# 在 Scala 中编写 AWS Glue ETL 脚本您可以在 GitHub 网站上的 [AWS Glue 示例存储库](https://github.com/awslabs/aws-glue-samples)中找到 AWS Glue 的 Scala 代码示例和实用程序。 AWS Glue 支持使用 PySpark Scala 方言的扩展来编写提取、转换和加载（ETL）任务脚本。下面几节介绍如何在 ETL 脚本中使用 AWS Glue Scala 库和 AWS Glue API，并提供了用于库的参考文档。 **Contents** + [使用 Scala](glue-etl-scala-using.md) + [在 DevEndpoint 笔记本上测试](glue-etl-scala-using.md#aws-glue-programming-scala-using-notebook) + [在 DevEndpoint REPL 中测试](glue-etl-scala-using.md#aws-glue-programming-scala-using-repl) + [Scala 脚本示例](glue-etl-scala-example.md) + [Scala API 列表](glue-etl-scala-apis.md) + [com.amazonaws.services.glue](glue-etl-scala-apis.md#glue-etl-scala-apis-glue) + [com.amazonaws.services.glue.ml](glue-etl-scala-apis.md#glue-etl-scala-apis-glue-ml) + [com.amazonaws.services.glue.dq](glue-etl-scala-apis.md#glue-etl-scala-apis-glue-dq) + [com.amazonaws.services.glue.types](glue-etl-scala-apis.md#glue-etl-scala-apis-glue-types) + [com.amazonaws.services.glue.util](glue-etl-scala-apis.md#glue-etl-scala-apis-glue-util) + [ChoiceOption](glue-etl-scala-apis-glue-choiceoption.md) + [ChoiceOption 特性](glue-etl-scala-apis-glue-choiceoption.md#glue-etl-scala-apis-glue-choiceoption-trait) + [ChoiceOption 对象](glue-etl-scala-apis-glue-choiceoption.md#glue-etl-scala-apis-glue-choiceoption-object) + [Apply](glue-etl-scala-apis-glue-choiceoption.md#glue-etl-scala-apis-glue-choiceoption-object-def-apply) + [ChoiceOptionWithResolver](glue-etl-scala-apis-glue-choiceoption.md#glue-etl-scala-apis-glue-choiceoptionwithresolver-case-class) + [MatchCatalogSchemaChoiceOption](glue-etl-scala-apis-glue-choiceoption.md#glue-etl-scala-apis-glue-matchcatalogschemachoiceoption-case-class) + [DataSink](glue-etl-scala-apis-glue-datasink-class.md) + [writeDynamicFrame](glue-etl-scala-apis-glue-datasink-class.md#glue-etl-scala-apis-glue-datasink-class-defs-writeDynamicFrame) + [pyWriteDynamicFrame](glue-etl-scala-apis-glue-datasink-class.md#glue-etl-scala-apis-glue-datasink-class-defs-pyWriteDynamicFrame) + [writeDataFrame](glue-etl-scala-apis-glue-datasink-class.md#glue-etl-scala-apis-glue-datasink-class-defs-writeDataFrame) + [pyWriteDataFrame](glue-etl-scala-apis-glue-datasink-class.md#glue-etl-scala-apis-glue-datasink-class-defs-pyWriteDataFrame) + [setCatalogInfo](glue-etl-scala-apis-glue-datasink-class.md#glue-etl-scala-apis-glue-datasink-class-defs-setCatalogInfo) + [supportsFormat](glue-etl-scala-apis-glue-datasink-class.md#glue-etl-scala-apis-glue-datasink-class-defs-supportsFormat) + [setFormat](glue-etl-scala-apis-glue-datasink-class.md#glue-etl-scala-apis-glue-datasink-class-defs-setFormat) + [withFormat](glue-etl-scala-apis-glue-datasink-class.md#glue-etl-scala-apis-glue-datasink-class-defs-withFormat) + [setAccumulableSize](glue-etl-scala-apis-glue-datasink-class.md#glue-etl-scala-apis-glue-datasink-class-defs-setAccumulableSize) + [getOutputErrorRecordsAccumulable](glue-etl-scala-apis-glue-datasink-class.md#glue-etl-scala-apis-glue-datasink-class-defs-getOutputErrorRecordsAccumulable) + [errorsAsDynamicFrame](glue-etl-scala-apis-glue-datasink-class.md#glue-etl-scala-apis-glue-datasink-class-defs-errorsAsDynamicFrame) + [DataSink 对象](glue-etl-scala-apis-glue-datasink-class.md#glue-etl-scala-apis-glue-datasink-object) + [recordMetrics](glue-etl-scala-apis-glue-datasink-class.md#glue-etl-scala-apis-glue-datasink-object-defs-recordMetrics) + [数据源特性](glue-etl-scala-apis-glue-datasource-trait.md) + [DynamicFrame](glue-etl-scala-apis-glue-dynamicframe.md) + [DynamicFrame 类](glue-etl-scala-apis-glue-dynamicframe-class.md) + [errorsCount](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-vals-errorsCount) + [applyMapping](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-applyMapping) + [assertErrorThreshold](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-assertErrorThreshold) + [计数](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-count) + [dropField](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-dropField) + [dropFields](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-dropFields) + [dropNulls](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-dropNulls) + [errorsAsDynamicFrame](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-errorsAsDynamicFrame) + [筛选条件](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-filter) + [getName](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-getName) + [getNumPartitions](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-getNumPartitions) + [getSchemaIfComputed](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-getSchemaIfComputed) + [isSchemaComputed](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-isSchemaComputed) + [javaToPython](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-javaToPython) + [联接](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-join) + [Map](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-map) + [mergeDynamicFrames](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-merge) + [printSchema](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-printSchema) + [recomputeSchema](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-recomputeSchema) + [Relationalize](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-relationalize) + [renameField](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-renameField) + [Repartition](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-repartition) + [resolveChoice](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-resolveChoice) + [架构](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-schema) + [selectField](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-selectField) + [selectFields](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-selectFields) + [显示](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-show) + [SimplifyDDBJson](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-simplifyDDBJson) + [Spigot](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-spigot) + [splitFields](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-splitFields) + [Def splitRows](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-splitRows) + [stageErrorsCount](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-stageErrorsCount) + [toDF](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-toDF) + [Unbox](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-unbox) + [Unnest](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-unnest) + [unnestDDBJson](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-unnestddbjson) + [withFrameSchema](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-withFrameSchema) + [Def withName](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-withName) + [withTransformationContext](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-withTransformationContext) + [DynamicFrame 对象](glue-etl-scala-apis-glue-dynamicframe-object.md) + [Def apply](glue-etl-scala-apis-glue-dynamicframe-object.md#glue-etl-scala-apis-glue-dynamicframe-object-defs-apply) + [Def emptyDynamicFrame](glue-etl-scala-apis-glue-dynamicframe-object.md#glue-etl-scala-apis-glue-dynamicframe-object-defs-emptyDynamicFrame) + [Def fromPythonRDD](glue-etl-scala-apis-glue-dynamicframe-object.md#glue-etl-scala-apis-glue-dynamicframe-object-defs-fromPythonRDD) + [Def ignoreErrors](glue-etl-scala-apis-glue-dynamicframe-object.md#glue-etl-scala-apis-glue-dynamicframe-object-defs-ignoreErrors) + [Def inlineErrors](glue-etl-scala-apis-glue-dynamicframe-object.md#glue-etl-scala-apis-glue-dynamicframe-object-defs-inlineErrors) + [Def newFrameWithErrors](glue-etl-scala-apis-glue-dynamicframe-object.md#glue-etl-scala-apis-glue-dynamicframe-object-defs-newFrameWithErrors) + [DynamicRecord](glue-etl-scala-apis-glue-dynamicrecord-class.md) + [addField](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-addField) + [dropField](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-dropField) + [setError](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-setError) + [isError](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-isError) + [getError](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-getError) + [clearError](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-clearError) + [写入](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-write) + [readFields](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-readFields) + [Clone](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-clone) + [架构](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-schema) + [getRoot](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-getRoot) + [toJson](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-toJson) + [getFieldNode](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-getFieldNode) + [getField](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-getField) + [hashCode](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-hashCode) + [Equals](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-class-defs-equals) + [DynamicRecord 对象](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-object) + [Apply](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-dynamicrecord-object-defs-apply) + [RecordTraverser 特性](glue-etl-scala-apis-glue-dynamicrecord-class.md#glue-etl-scala-apis-glue-recordtraverser-trait) + [GlueContext](glue-etl-scala-apis-glue-gluecontext.md) + [addIngestionTimeColumns](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-addIngestionTimeColumns) + [createDataFrameFromOptions](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-createDataFrameFromOptions) + [forEachBatch](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-forEachBatch) + [getCatalogSink](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getCatalogSink) + [getCatalogSource](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getCatalogSource) + [getJDBCSink](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getJDBCSink) + [getSink](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSink) + [getSinkWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSinkWithFormat) + [getSource](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSource) + [getSourceWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSourceWithFormat) + [getSparkSession](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSparkSession) + [startTransaction](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-start-transaction) + [commitTransaction](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-commit-transaction) + [cancelTransaction](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-cancel-transaction) + [this](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-this-1) + [this](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-this-2) + [this](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-this-3) + [MappingSpec](glue-etl-scala-apis-glue-mappingspec.md) + [MappingSpec case 类](glue-etl-scala-apis-glue-mappingspec.md#glue-etl-scala-apis-glue-mappingspec-case-class) + [MappingSpec 对象](glue-etl-scala-apis-glue-mappingspec.md#glue-etl-scala-apis-glue-mappingspec-object) + [orderingByTarget](glue-etl-scala-apis-glue-mappingspec.md#glue-etl-scala-apis-glue-mappingspec-object-val-orderingbytarget) + [Apply](glue-etl-scala-apis-glue-mappingspec.md#glue-etl-scala-apis-glue-mappingspec-object-defs-apply-1) + [Apply](glue-etl-scala-apis-glue-mappingspec.md#glue-etl-scala-apis-glue-mappingspec-object-defs-apply-2) + [Apply](glue-etl-scala-apis-glue-mappingspec.md#glue-etl-scala-apis-glue-mappingspec-object-defs-apply-3) + [ResolveSpec](glue-etl-scala-apis-glue-resolvespec.md) + [ResolveSpec 对象](glue-etl-scala-apis-glue-resolvespec.md#glue-etl-scala-apis-glue-resolvespec-object) + [Def](glue-etl-scala-apis-glue-resolvespec.md#glue-etl-scala-apis-glue-resolvespec-object-def-apply_1) + [Def](glue-etl-scala-apis-glue-resolvespec.md#glue-etl-scala-apis-glue-resolvespec-object-def-apply_2) + [ResolveSpec case 类](glue-etl-scala-apis-glue-resolvespec.md#glue-etl-scala-apis-glue-resolvespec-case-class) + [Def 方法](glue-etl-scala-apis-glue-resolvespec.md#glue-etl-scala-apis-glue-resolvespec-case-class-defs) + [ArrayNode](glue-etl-scala-apis-glue-types-arraynode.md) + [ArrayNode case 类](glue-etl-scala-apis-glue-types-arraynode.md#glue-etl-scala-apis-glue-types-arraynode-case-class) + [Def 方法](glue-etl-scala-apis-glue-types-arraynode.md#glue-etl-scala-apis-glue-types-arraynode-case-class-defs) + [BinaryNode](glue-etl-scala-apis-glue-types-binarynode.md) + [BinaryNode case 类](glue-etl-scala-apis-glue-types-binarynode.md#glue-etl-scala-apis-glue-types-binarynode-case-class) + [Val 字段](glue-etl-scala-apis-glue-types-binarynode.md#glue-etl-scala-apis-glue-types-binarynode-case-class-vals) + [Def 方法](glue-etl-scala-apis-glue-types-binarynode.md#glue-etl-scala-apis-glue-types-binarynode-case-class-defs) + [BooleanNode](glue-etl-scala-apis-glue-types-booleannode.md) + [BooleanNode case 类](glue-etl-scala-apis-glue-types-booleannode.md#glue-etl-scala-apis-glue-types-booleannode-case-class) + [Val 字段](glue-etl-scala-apis-glue-types-booleannode.md#glue-etl-scala-apis-glue-types-booleannode-case-class-vals) + [Def 方法](glue-etl-scala-apis-glue-types-booleannode.md#glue-etl-scala-apis-glue-types-booleannode-case-class-defs) + [ByteNode](glue-etl-scala-apis-glue-types-bytenode.md) + [ByteNode case 类](glue-etl-scala-apis-glue-types-bytenode.md#glue-etl-scala-apis-glue-types-bytenode-case-class) + [Val 字段](glue-etl-scala-apis-glue-types-bytenode.md#glue-etl-scala-apis-glue-types-bytenode-case-class-vals) + [Def 方法](glue-etl-scala-apis-glue-types-bytenode.md#glue-etl-scala-apis-glue-types-bytenode-case-class-defs) + [DateNode](glue-etl-scala-apis-glue-types-datenode.md) + [DateNode case 类](glue-etl-scala-apis-glue-types-datenode.md#glue-etl-scala-apis-glue-types-datenode-case-class) + [Val 字段](glue-etl-scala-apis-glue-types-datenode.md#glue-etl-scala-apis-glue-types-datenode-case-class-vals) + [Def 方法](glue-etl-scala-apis-glue-types-datenode.md#glue-etl-scala-apis-glue-types-datenode-case-class-defs) + [DecimalNode](glue-etl-scala-apis-glue-types-decimalnode.md) + [DecimalNode case 类](glue-etl-scala-apis-glue-types-decimalnode.md#glue-etl-scala-apis-glue-types-decimalnode-case-class) + [Val 字段](glue-etl-scala-apis-glue-types-decimalnode.md#glue-etl-scala-apis-glue-types-decimalnode-case-class-vals) + [Def 方法](glue-etl-scala-apis-glue-types-decimalnode.md#glue-etl-scala-apis-glue-types-decimalnode-case-class-defs) + [DoubleNode](glue-etl-scala-apis-glue-types-doublenode.md) + [DoubleNode case 类](glue-etl-scala-apis-glue-types-doublenode.md#glue-etl-scala-apis-glue-types-doublenode-case-class) + [Val 字段](glue-etl-scala-apis-glue-types-doublenode.md#glue-etl-scala-apis-glue-types-doublenode-case-class-vals) + [Def 方法](glue-etl-scala-apis-glue-types-doublenode.md#glue-etl-scala-apis-glue-types-doublenode-case-class-defs) + [DynamicNode](glue-etl-scala-apis-glue-types-dynamicnode.md) + [DynamicNode 类](glue-etl-scala-apis-glue-types-dynamicnode.md#glue-etl-scala-apis-glue-types-dynamicnode-class) + [Def 方法](glue-etl-scala-apis-glue-types-dynamicnode.md#glue-etl-scala-apis-glue-types-dynamicnode-class-defs) + [DynamicNode 对象](glue-etl-scala-apis-glue-types-dynamicnode.md#glue-etl-scala-apis-glue-types-dynamicnode-object) + [Def 方法](glue-etl-scala-apis-glue-types-dynamicnode.md#glue-etl-scala-apis-glue-types-dynamicnode-object-defs) + [EvaluateDataQuality](glue-etl-scala-apis-glue-dq-EvaluateDataQuality.md) + [apply](glue-etl-scala-apis-glue-dq-EvaluateDataQuality.md#glue-etl-scala-apis-glue-dq-EvaluateDataQuality-defs-apply) + [示例](glue-etl-scala-apis-glue-dq-EvaluateDataQuality.md#glue-etl-scala-apis-glue-dq-EvaluateDataQuality-example) + [FloatNode](glue-etl-scala-apis-glue-types-floatnode.md) + [FloatNode case 类](glue-etl-scala-apis-glue-types-floatnode.md#glue-etl-scala-apis-glue-types-floatnode-case-class) + [Val 字段](glue-etl-scala-apis-glue-types-floatnode.md#glue-etl-scala-apis-glue-types-floatnode-case-class-vals) + [Def 方法](glue-etl-scala-apis-glue-types-floatnode.md#glue-etl-scala-apis-glue-types-floatnode-case-class-defs) + [FillMissingValues](glue-etl-scala-apis-glue-ml-fillmissingvalues.md) + [Apply](glue-etl-scala-apis-glue-ml-fillmissingvalues.md#glue-etl-scala-apis-glue-ml-fillmissingvalues-defs-apply) + [FindMatches](glue-etl-scala-apis-glue-ml-findmatches.md) + [Apply](glue-etl-scala-apis-glue-ml-findmatches.md#glue-etl-scala-apis-glue-ml-findmatches-defs-apply) + [FindIncrementalMatches](glue-etl-scala-apis-glue-ml-findincrementalmatches.md) + [Apply](glue-etl-scala-apis-glue-ml-findincrementalmatches.md#glue-etl-scala-apis-glue-ml-findincrementalmatches-defs-apply) + [IntegerNode](glue-etl-scala-apis-glue-types-integernode.md) + [IntegerNode case 类](glue-etl-scala-apis-glue-types-integernode.md#glue-etl-scala-apis-glue-types-integernode-case-class) + [Val 字段](glue-etl-scala-apis-glue-types-integernode.md#glue-etl-scala-apis-glue-types-integernode-case-class-vals) + [Def 方法](glue-etl-scala-apis-glue-types-integernode.md#glue-etl-scala-apis-glue-types-integernode-case-class-defs) + [LongNode](glue-etl-scala-apis-glue-types-longnode.md) + [LongNode case 类](glue-etl-scala-apis-glue-types-longnode.md#glue-etl-scala-apis-glue-types-longnode-case-class) + [Val 字段](glue-etl-scala-apis-glue-types-longnode.md#glue-etl-scala-apis-glue-types-longnode-case-class-vals) + [Def 方法](glue-etl-scala-apis-glue-types-longnode.md#glue-etl-scala-apis-glue-types-longnode-case-class-defs) + [MapLikeNode](glue-etl-scala-apis-glue-types-maplikenode.md) + [MapLikeNode 类](glue-etl-scala-apis-glue-types-maplikenode.md#glue-etl-scala-apis-glue-types-maplikenode-class) + [Def 方法](glue-etl-scala-apis-glue-types-maplikenode.md#glue-etl-scala-apis-glue-types-maplikenode-class-defs) + [MapNode](glue-etl-scala-apis-glue-types-mapnode.md) + [MapNode case 类](glue-etl-scala-apis-glue-types-mapnode.md#glue-etl-scala-apis-glue-types-mapnode-case-class) + [Def 方法](glue-etl-scala-apis-glue-types-mapnode.md#glue-etl-scala-apis-glue-types-mapnode-case-class-defs) + [NullNode](glue-etl-scala-apis-glue-types-nullnode.md) + [NullNode 类](glue-etl-scala-apis-glue-types-nullnode.md#glue-etl-scala-apis-glue-types-nullnode-class) + [NullNode case 对象](glue-etl-scala-apis-glue-types-nullnode.md#glue-etl-scala-apis-glue-types-nullnode-case-object) + [ObjectNode](glue-etl-scala-apis-glue-types-objectnode.md) + [ObjectNode 对象](glue-etl-scala-apis-glue-types-objectnode.md#glue-etl-scala-apis-glue-types-objectnode-object) + [Def 方法](glue-etl-scala-apis-glue-types-objectnode.md#glue-etl-scala-apis-glue-types-objectnode-object-defs) + [ObjectNode case 类](glue-etl-scala-apis-glue-types-objectnode.md#glue-etl-scala-apis-glue-types-objectnode-case-class) + [Def 方法](glue-etl-scala-apis-glue-types-objectnode.md#glue-etl-scala-apis-glue-types-objectnode-case-class-defs) + [ScalarNode](glue-etl-scala-apis-glue-types-scalarnode.md) + [ScalarNode 类](glue-etl-scala-apis-glue-types-scalarnode.md#glue-etl-scala-apis-glue-types-scalarnode-class) + [Def 方法](glue-etl-scala-apis-glue-types-scalarnode.md#glue-etl-scala-apis-glue-types-scalarnode-class-defs) + [ScalarNode 对象](glue-etl-scala-apis-glue-types-scalarnode.md#glue-etl-scala-apis-glue-types-scalarnode-object) + [Def 方法](glue-etl-scala-apis-glue-types-scalarnode.md#glue-etl-scala-apis-glue-types-scalarnode-object-defs) + [ShortNode](glue-etl-scala-apis-glue-types-shortnode.md) + [ShortNode case 类](glue-etl-scala-apis-glue-types-shortnode.md#glue-etl-scala-apis-glue-types-shortnode-case-class) + [Val 字段](glue-etl-scala-apis-glue-types-shortnode.md#glue-etl-scala-apis-glue-types-shortnode-case-class-vals) + [Def 方法](glue-etl-scala-apis-glue-types-shortnode.md#glue-etl-scala-apis-glue-types-shortnode-case-class-defs) + [StringNode](glue-etl-scala-apis-glue-types-stringnode.md) + [StringNode case 类](glue-etl-scala-apis-glue-types-stringnode.md#glue-etl-scala-apis-glue-types-stringnode-case-class) + [Val 字段](glue-etl-scala-apis-glue-types-stringnode.md#glue-etl-scala-apis-glue-types-stringnode-case-class-vals) + [Def 方法](glue-etl-scala-apis-glue-types-stringnode.md#glue-etl-scala-apis-glue-types-stringnode-case-class-defs) + [TimestampNode](glue-etl-scala-apis-glue-types-timestampnode.md) + [TimestampNode case 类](glue-etl-scala-apis-glue-types-timestampnode.md#glue-etl-scala-apis-glue-types-timestampnode-case-class) + [Val 字段](glue-etl-scala-apis-glue-types-timestampnode.md#glue-etl-scala-apis-glue-types-timestampnode-case-class-vals) + [Def 方法](glue-etl-scala-apis-glue-types-timestampnode.md#glue-etl-scala-apis-glue-types-timestampnode-case-class-defs) + [GlueArgParser](glue-etl-scala-apis-glue-util-glueargparser.md) + [GlueArgParser 对象](glue-etl-scala-apis-glue-util-glueargparser.md#glue-etl-scala-apis-glue-util-glueargparser-object) + [Def 方法](glue-etl-scala-apis-glue-util-glueargparser.md#glue-etl-scala-apis-glue-util-glueargparser-object-defs) + [作业](glue-etl-scala-apis-glue-util-job.md) + [Job 对象](glue-etl-scala-apis-glue-util-job.md#glue-etl-scala-apis-glue-util-job-object) + [Def 方法](glue-etl-scala-apis-glue-util-job.md#glue-etl-scala-apis-glue-util-job-object-defs) # 使用 Scala 编写 AWS Glue ETL 脚本您可以使用 AWS Glue 控制台自动生成 Scala 提取、转换和加载 (ETL) 程序，并根据需要对其进行修改，然后将其分配给作业。或者，您可以从头编写自己的程序。有关更多信息，请参阅[在 AWS Glue 中为 Spark 作业配置作业属性](add-job.md)。然后，AWS Glue 会在服务器上编译您的 Scala 程序，再运行关联的作业。为了确保您的程序编译时不会出现错误并可按预期运行，请务必在 REPL（读取-求值-输出-循环）或 Jupyter Notebook 的开发端点中加载程序并在此处进行测试，然后再在作业中运行。由于编译过程在服务器上进行，因此您无法清楚地看到服务器上出现的任何问题。 ## 在 Jupyter Notebook 中的开发端点上测试 Scala ETL 程序要在 AWS Glue 开发终端节点上测试 Scala 程序，请设置开发终端节点（如 [添加开发终端节点](add-dev-endpoint.md) 中所述）。接下来，将其连接到在您的计算机上本地运行或在 Amazon EC2 笔记本服务器上远程运行的 Jupyter Notebook。要安装 Jupyter Notebook 的本地版本，请按照 [教程：JupyterLab 中的 Jupyter notebook](dev-endpoint-tutorial-local-jupyter.md) 中的说明操作。在您的笔记本上运行 Scala 代码与运行 PySpark 代码之间的唯一差别在于，在笔记本上每个段落的开头必须为： ``` %spark ``` 这将防止 Notebook 服务器默认使用 Spark 解释器的 PySpark 风格。 ## 在 Scala REPL 中测试 Scala ETL 程序您可以使用 AWS Glue Scala REPL 在开发终端节点上测试 Scala 程序。按照[教程：使用 SageMaker AI 笔记本教程：使用 REPL shell](dev-endpoint-tutorial-repl.md)中的说明操作，除了 SSH-to-REPL 命令末尾，将 `-t gluepyspark` 替换为 `-t glue-spark-shell`。这将调用 AWS Glue Scala REPL。要在完成时关闭 REPL，请键入 `sys.exit`。 # Scala 脚本示例 - 流式处理 ETL **Example** 以下示例脚本连接到 Amazon Kinesis Data Streams，使用来自数据目录的架构解析数据流，将流联接到 Amazon S3 上的静态数据集，并以 parquet 格式将联接结果输出到 Amazon S3。 ``` // This script connects to an Amazon Kinesis stream, uses a schema from the data catalog to parse the stream, // joins the stream to a static dataset on Amazon S3, and outputs the joined results to Amazon S3 in parquet format. import com.amazonaws.services.glue.GlueContext import com.amazonaws.services.glue.util.GlueArgParser import com.amazonaws.services.glue.util.Job import java.util.Calendar import org.apache.spark.SparkContext import org.apache.spark.sql.Dataset import org.apache.spark.sql.Row import org.apache.spark.sql.SaveMode import org.apache.spark.sql.SparkSession import org.apache.spark.sql.functions.from_json import org.apache.spark.sql.streaming.Trigger import scala.collection.JavaConverters._ object streamJoiner { def main(sysArgs: Array[String]) { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) val sparkSession: SparkSession = glueContext.getSparkSession import sparkSession.implicits._ // @params: [JOB_NAME] val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray) Job.init(args("JOB_NAME"), glueContext, args.asJava) val staticData = sparkSession.read // read() returns type DataFrameReader .format("csv") .option("header", "true") .load("s3://amzn-s3-demo-bucket/inputs/productsStatic.csv") // load() returns a DataFrame val datasource0 = sparkSession.readStream // readstream() returns type DataStreamReader .format("kinesis") .option("streamName", "stream-join-demo") .option("endpointUrl", "https://kinesis.us-east-1.amazonaws.com") .option("startingPosition", "TRIM_HORIZON") .load // load() returns a DataFrame val selectfields1 = datasource0.select(from_json($"data".cast("string"), glueContext.getCatalogSchemaAsSparkSchema("stream-demos", "stream-join-demo2")) as "data").select("data.*") val datasink2 = selectfields1.writeStream.foreachBatch { (dataFrame: Dataset[Row], batchId: Long) => { //foreachBatch() returns type DataStreamWriter val joined = dataFrame.join(staticData, "product_id") val year: Int = Calendar.getInstance().get(Calendar.YEAR) val month :Int = Calendar.getInstance().get(Calendar.MONTH) + 1 val day: Int = Calendar.getInstance().get(Calendar.DATE) val hour: Int = Calendar.getInstance().get(Calendar.HOUR_OF_DAY) if (dataFrame.count() > 0) { joined.write // joined.write returns type DataFrameWriter .mode(SaveMode.Append) .format("parquet") .option("quote", " ") .save("s3://amzn-s3-demo-bucket/output/" + "/year=" + "%04d".format(year) + "/month=" + "%02d".format(month) + "/day=" + "%02d".format(day) + "/hour=" + "%02d".format(hour) + "/") } } } // end foreachBatch() .trigger(Trigger.ProcessingTime("100 seconds")) .option("checkpointLocation", "s3://amzn-s3-demo-bucket/checkpoint/") .start().awaitTermination() // start() returns type StreamingQuery Job.commit() } } ``` # AWS Glue Scala 库中的 API AWS Glue 支持使用 PySpark Scala 方言的扩展来编写提取、转换和加载 (ETL) 作业脚本。以下部分描述了 AWS Glue Scala 库中的 API。 ## com.amazonaws.services.glue ** Scala 库中的 **com.amazonaws.services.glueAWS Glue 程序包包含以下 API： + [ChoiceOption](glue-etl-scala-apis-glue-choiceoption.md) + [DataSink](glue-etl-scala-apis-glue-datasink-class.md) + [数据源特性](glue-etl-scala-apis-glue-datasource-trait.md) + [DynamicFrame](glue-etl-scala-apis-glue-dynamicframe.md) + [DynamicRecord](glue-etl-scala-apis-glue-dynamicrecord-class.md) + [GlueContext](glue-etl-scala-apis-glue-gluecontext.md) + [MappingSpec](glue-etl-scala-apis-glue-mappingspec.md) + [ResolveSpec](glue-etl-scala-apis-glue-resolvespec.md) ## com.amazonaws.services.glue.ml AWS Glue Scala 库中的 **com.amazonaws.services.glue.ml** 程序包包含以下 API： + [FillMissingValues](glue-etl-scala-apis-glue-ml-fillmissingvalues.md) + [FindIncrementalMatches](glue-etl-scala-apis-glue-ml-findincrementalmatches.md) + [FindMatches](glue-etl-scala-apis-glue-ml-findmatches.md) ## com.amazonaws.services.glue.dq AWS Glue Scala 库中的 **com.amazonaws.services.glue.dq** 程序包包含以下 API： + [EvaluateDataQuality](glue-etl-scala-apis-glue-dq-EvaluateDataQuality.md) ## com.amazonaws.services.glue.types ** Scala 库中的 **com.amazonaws.services.glue.typesAWS Glue 程序包包含以下 API： + [ArrayNode](glue-etl-scala-apis-glue-types-arraynode.md) + [BinaryNode](glue-etl-scala-apis-glue-types-binarynode.md) + [BooleanNode](glue-etl-scala-apis-glue-types-booleannode.md) + [ByteNode](glue-etl-scala-apis-glue-types-bytenode.md) + [DateNode](glue-etl-scala-apis-glue-types-datenode.md) + [DecimalNode](glue-etl-scala-apis-glue-types-decimalnode.md) + [DoubleNode](glue-etl-scala-apis-glue-types-doublenode.md) + [DynamicNode](glue-etl-scala-apis-glue-types-dynamicnode.md) + [FloatNode](glue-etl-scala-apis-glue-types-floatnode.md) + [IntegerNode](glue-etl-scala-apis-glue-types-integernode.md) + [LongNode](glue-etl-scala-apis-glue-types-longnode.md) + [MapLikeNode](glue-etl-scala-apis-glue-types-maplikenode.md) + [MapNode](glue-etl-scala-apis-glue-types-mapnode.md) + [NullNode](glue-etl-scala-apis-glue-types-nullnode.md) + [ObjectNode](glue-etl-scala-apis-glue-types-objectnode.md) + [ScalarNode](glue-etl-scala-apis-glue-types-scalarnode.md) + [ShortNode](glue-etl-scala-apis-glue-types-shortnode.md) + [StringNode](glue-etl-scala-apis-glue-types-stringnode.md) + [TimestampNode](glue-etl-scala-apis-glue-types-timestampnode.md) ## com.amazonaws.services.glue.util ** Scala 库中的 **com.amazonaws.services.glue.utilAWS Glue 程序包包含以下 API： + [GlueArgParser](glue-etl-scala-apis-glue-util-glueargparser.md) + [作业](glue-etl-scala-apis-glue-util-job.md) # AWS Glue Scala ChoiceOption API **Topics** + [ChoiceOption 特性](#glue-etl-scala-apis-glue-choiceoption-trait) + [ChoiceOption 对象](#glue-etl-scala-apis-glue-choiceoption-object) + [Case 类 ChoiceOptionWithResolver](#glue-etl-scala-apis-glue-choiceoptionwithresolver-case-class) + [Case 类 MatchCatalogSchemaChoiceOption](#glue-etl-scala-apis-glue-matchcatalogschemachoiceoption-case-class) **程序包：com.amazonaws.services.glue** ## ChoiceOption 特性 ``` trait ChoiceOption extends Serializable ``` ## ChoiceOption 对象 **ChoiceOption** ``` object ChoiceOption ``` 用于解决适用于 `DynamicFrame` 中所有 `ChoiceType` 节点的选项问题的一般策略。 + `val CAST` + `val MAKE_COLS` + `val MAKE_STRUCT` + `val MATCH_CATALOG` + `val PROJECT` ### Def apply ``` def apply(choice: String): ChoiceOption ``` ## Case 类 ChoiceOptionWithResolver ``` case class ChoiceOptionWithResolver(name: String, choiceResolver: ChoiceResolver) extends ChoiceOption {} ``` ## Case 类 MatchCatalogSchemaChoiceOption ``` case class MatchCatalogSchemaChoiceOption() extends ChoiceOption {} ``` # 抽象 DataSink 类 **Topics** + [Def writeDynamicFrame](#glue-etl-scala-apis-glue-datasink-class-defs-writeDynamicFrame) + [Def pyWriteDynamicFrame](#glue-etl-scala-apis-glue-datasink-class-defs-pyWriteDynamicFrame) + [Def writeDataFrame](#glue-etl-scala-apis-glue-datasink-class-defs-writeDataFrame) + [Def pyWriteDataFrame](#glue-etl-scala-apis-glue-datasink-class-defs-pyWriteDataFrame) + [Def setCatalogInfo](#glue-etl-scala-apis-glue-datasink-class-defs-setCatalogInfo) + [Def supportsFormat](#glue-etl-scala-apis-glue-datasink-class-defs-supportsFormat) + [Def setFormat](#glue-etl-scala-apis-glue-datasink-class-defs-setFormat) + [Def withFormat](#glue-etl-scala-apis-glue-datasink-class-defs-withFormat) + [Def setAccumulableSize](#glue-etl-scala-apis-glue-datasink-class-defs-setAccumulableSize) + [Def getOutputErrorRecordsAccumulable](#glue-etl-scala-apis-glue-datasink-class-defs-getOutputErrorRecordsAccumulable) + [Def errorsAsDynamicFrame](#glue-etl-scala-apis-glue-datasink-class-defs-errorsAsDynamicFrame) + [DataSink 对象](#glue-etl-scala-apis-glue-datasink-object) **程序包：com.amazonaws.services.glue** ``` abstract class DataSink ``` `DataSource` 的写入器模拟。`DataSink` 封装可将 `DynamicFrame` 写入到的目标和格式。 ## Def writeDynamicFrame ``` def writeDynamicFrame( frame : DynamicFrame, callSite : CallSite = CallSite("Not provided", "") ) : DynamicFrame ``` ## Def pyWriteDynamicFrame ``` def pyWriteDynamicFrame( frame : DynamicFrame, site : String = "Not provided", info : String = "" ) ``` ## Def writeDataFrame ``` def writeDataFrame(frame: DataFrame, glueContext: GlueContext, callSite: CallSite = CallSite("Not provided", "") ): DataFrame ``` ## Def pyWriteDataFrame ``` def pyWriteDataFrame(frame: DataFrame, glueContext: GlueContext, site: String = "Not provided", info: String = "" ): DataFrame ``` ## Def setCatalogInfo ``` def setCatalogInfo(catalogDatabase: String, catalogTableName : String, catalogId : String = "") ``` ## Def supportsFormat ``` def supportsFormat( format : String ) : Boolean ``` ## Def setFormat ``` def setFormat( format : String, options : JsonOptions ) : Unit ``` ## Def withFormat ``` def withFormat( format : String, options : JsonOptions = JsonOptions.empty ) : DataSink ``` ## Def setAccumulableSize ``` def setAccumulableSize( size : Int ) : Unit ``` ## Def getOutputErrorRecordsAccumulable ``` def getOutputErrorRecordsAccumulable : Accumulable[List[OutputError], OutputError] ``` ## Def errorsAsDynamicFrame ``` def errorsAsDynamicFrame : DynamicFrame ``` ## DataSink 对象 ``` object DataSink ``` ### Def recordMetrics ``` def recordMetrics( frame : DynamicFrame, ctxt : String ) : DynamicFrame ``` # AWS Glue Scala 数据源特性 **程序包：com.amazonaws.services.glue** 用于生成 `DynamicFrame` 的高级接口。 ``` trait DataSource { def getDynamicFrame : DynamicFrame def getDynamicFrame( minPartitions : Int, targetPartitions : Int ) : DynamicFrame def getDataFrame : DataFrame /** @param num: the number of records for sampling. * @param options: optional parameters to control sampling behavior. Current available parameter for Amazon S3 sources in options: * 1. maxSamplePartitions: the maximum number of partitions the sampling will read. * 2. maxSampleFilesPerPartition: the maximum number of files the sampling will read in one partition. */ def getSampleDynamicFrame(num:Int, options: JsonOptions = JsonOptions.empty): DynamicFrame def glueContext : GlueContext def setFormat( format : String, options : String ) : Unit def setFormat( format : String, options : JsonOptions ) : Unit def supportsFormat( format : String ) : Boolean def withFormat( format : String, options : JsonOptions = JsonOptions.empty ) : DataSource } ``` # AWS Glue Scala DynamicFrame API **程序包：com.amazonaws.services.glue** **Contents** + [AWS Glue Scala DynamicFrame 类](glue-etl-scala-apis-glue-dynamicframe-class.md) + [Val errorsCount](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-vals-errorsCount) + [Def applyMapping](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-applyMapping) + [Def assertErrorThreshold](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-assertErrorThreshold) + [Def count](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-count) + [Def dropField](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-dropField) + [Def dropFields](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-dropFields) + [Def dropNulls](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-dropNulls) + [Def errorsAsDynamicFrame](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-errorsAsDynamicFrame) + [Def filter](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-filter) + [Def getName](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-getName) + [Def getNumPartitions](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-getNumPartitions) + [Def getSchemaIfComputed](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-getSchemaIfComputed) + [Def isSchemaComputed](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-isSchemaComputed) + [Def javaToPython](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-javaToPython) + [Def join](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-join) + [Def map](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-map) + [Def mergeDynamicFrames](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-merge) + [Def printSchema](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-printSchema) + [Def recomputeSchema](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-recomputeSchema) + [Def relationalize](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-relationalize) + [Def renameField](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-renameField) + [Def repartition](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-repartition) + [Def resolveChoice](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-resolveChoice) + [Def schema](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-schema) + [Def selectField](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-selectField) + [Def selectFields](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-selectFields) + [Def show](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-show) + [Def simplifyDDBJson](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-simplifyDDBJson) + [Def spigot](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-spigot) + [Def splitFields](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-splitFields) + [Def splitRows](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-splitRows) + [Def stageErrorsCount](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-stageErrorsCount) + [Def toDF](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-toDF) + [Def unbox](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-unbox) + [Def unnest](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-unnest) + [Def unnestDDBJson](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-unnestddbjson) + [Def withFrameSchema](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-withFrameSchema) + [Def withName](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-withName) + [Def withTransformationContext](glue-etl-scala-apis-glue-dynamicframe-class.md#glue-etl-scala-apis-glue-dynamicframe-class-defs-withTransformationContext) + [DynamicFrame 对象](glue-etl-scala-apis-glue-dynamicframe-object.md) + [Def apply](glue-etl-scala-apis-glue-dynamicframe-object.md#glue-etl-scala-apis-glue-dynamicframe-object-defs-apply) + [Def emptyDynamicFrame](glue-etl-scala-apis-glue-dynamicframe-object.md#glue-etl-scala-apis-glue-dynamicframe-object-defs-emptyDynamicFrame) + [Def fromPythonRDD](glue-etl-scala-apis-glue-dynamicframe-object.md#glue-etl-scala-apis-glue-dynamicframe-object-defs-fromPythonRDD) + [Def ignoreErrors](glue-etl-scala-apis-glue-dynamicframe-object.md#glue-etl-scala-apis-glue-dynamicframe-object-defs-ignoreErrors) + [Def inlineErrors](glue-etl-scala-apis-glue-dynamicframe-object.md#glue-etl-scala-apis-glue-dynamicframe-object-defs-inlineErrors) + [Def newFrameWithErrors](glue-etl-scala-apis-glue-dynamicframe-object.md#glue-etl-scala-apis-glue-dynamicframe-object-defs-newFrameWithErrors) # AWS Glue Scala DynamicFrame 类 **程序包：com.amazonaws.services.glue** ``` class DynamicFrame extends Serializable with Logging ( val glueContext : GlueContext, _records : RDD[DynamicRecord], val name : String = s"", val transformationContext : String = DynamicFrame.UNDEFINED, callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0, prevErrors : => Long = 0, errorExpr : => Unit = {} ) ``` `DynamicFrame` 是自描述的 [DynamicRecord](glue-etl-scala-apis-glue-dynamicrecord-class.md) 对象的分布式集合。 `DynamicFrame` 旨在为 ETL（提取、转换和加载）操作提供灵活的数据模型。它们不需要创建架构，可用于读取和转换具有杂乱或不一致的值和类型的数据。可以按需为需要架构的那些操作计算架构。 `DynamicFrame` 为数据清理和 ETL 提供了一系列转换。它们还支持转换为 SparkSQL DataFrame 和从其转换以与现有代码和 DataFrame 提供的许多分析操作集成。跨构造 `DynamicFrame` 的许多 AWS Glue 转换共享以下参数： + `transformationContext` — 此 `DynamicFrame` 的标识符。`transformationContext` 用作跨运行保存的作业书签状态的密钥。 + `callSite` – 为错误报告提供上下文信息。在从 Python 调用时，会自动设置这些值。 + `stageThreshold` – 在引发异常之前允许的来自此 `DynamicFrame` 计算的最大错误记录数，不包括以前的 `DynamicFrame` 中存在的记录。 + `totalThreshold` – 引发异常之前的最大错误记录总数，包括以前的帧中的记录。 ## Val errorsCount ``` val errorsCount ``` 此 `DynamicFrame` 中的错误记录数。这包括来自以前的操作的错误。 ## Def applyMapping ``` def applyMapping( mappings : Seq[Product4[String, String, String, String]], caseSensitive : Boolean = true, transformationContext : String = "", callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` + `mappings` – 用于构造新 `DynamicFrame` 的映射序列。 + `caseSensitive` – 是否将源列视为区分大小写。在与不区分大小写的存储（如 AWS Glue 数据目录）集成时，将此项设置为 false 可能很有帮助。基于一系列的映射选择、投影和转换列。每个映射由源列和类型以及目标列和类型构成。映射可指定为四元组 (`source_path`、`source_type`、` target_path`、`target_type`) 或包含相同信息的 [MappingSpec](glue-etl-scala-apis-glue-mappingspec.md) 对象。除了将映射用于简单的投影和转换外，还可以通过使用“`.`”（句点）分隔路径的组件来用于嵌套或取消嵌套字段。例如，假设您有一个包含以下架构的 `DynamicFrame`。 ``` {{{ root |-- name: string |-- age: int |-- address: struct | |-- state: string | |-- zip: int }}} ``` 您可以进行以下调用来取消嵌套 `state` 和 `zip` 字段。 ``` {{{ df.applyMapping( Seq(("name", "string", "name", "string"), ("age", "int", "age", "int"), ("address.state", "string", "state", "string"), ("address.zip", "int", "zip", "int"))) }}} ``` 生成的架构如下所示。 ``` {{{ root |-- name: string |-- age: int |-- state: string |-- zip: int }}} ``` 您还可以使用 `applyMapping` 来重新嵌套列。例如，以下代码反转以前的转换，并在目标中创建一个名为 `address` 的结构。 ``` {{{ df.applyMapping( Seq(("name", "string", "name", "string"), ("age", "int", "age", "int"), ("state", "string", "address.state", "string"), ("zip", "int", "address.zip", "int"))) }}} ``` 包含“`.`”（句点）字符的字段名称可以使用反引号 (````) 括起来。 **注意** 目前，您不能使用 `applyMapping` 方法映射嵌套在数组下的列。 ## Def assertErrorThreshold ``` def assertErrorThreshold : Unit ``` 强制计算并验证错误记录数是否低于 `stageThreshold` 和 `totalThreshold` 的操作。如果任一条件失败，则引发异常。 ## Def count ``` lazy def count ``` 返回 `DynamicFrame` 中的元素数量。 ## Def dropField ``` def dropField( path : String, transformationContext : String = "", callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` 返回已删除指定列的新 `DynamicFrame`。 ## Def dropFields ``` def dropFields( fieldNames : Seq[String], // The column names to drop. transformationContext : String = "", callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` 返回已删除指定列的新 `DynamicFrame`。您可以使用此方法删除嵌套列（包括数组中的列），但不能用于删除特定数组元素。 ## Def dropNulls ``` def dropNulls( transformationContext : String = "", callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0 ) ``` 返回已删除所有空列的新 `DynamicFrame`。 **注意** 这只删除类型为 `NullType` 的列。不删除或修改其他列中的单个空值。 ## Def errorsAsDynamicFrame ``` def errorsAsDynamicFrame ``` 返回包含此 `DynamicFrame` 中的错误记录的新 `DynamicFrame`。 ## Def filter ``` def filter( f : DynamicRecord => Boolean, errorMsg : String = "", transformationContext : String = "", callSite : CallSite = CallSite("Not provided"), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` 构造只包含函数“`f`”为其返回 `true` 的那些记录的新 `DynamicFrame`。筛选器函数“`f`”不应转变输入记录。 ## Def getName ``` def getName : String ``` 返回此 `DynamicFrame` 的名称。 ## Def getNumPartitions ``` def getNumPartitions ``` 返回 `DynamicFrame` 中的分区数量。 ## Def getSchemaIfComputed ``` def getSchemaIfComputed : Option[Schema] ``` 如果架构已经过计算，则返回该架构。如果架构尚未经过计算，则不扫描数据。 ## Def isSchemaComputed ``` def isSchemaComputed : Boolean ``` 如果已为此 `DynamicFrame` 计算架构，则返回 `true`；否则返回 `false`。如果此方法返回 false，则调用 `schema` 方法将需要再次扫描此 `DynamicFrame` 中的记录。 ## Def javaToPython ``` def javaToPython : JavaRDD[Array[Byte]] ``` ## Def join ``` def join( keys1 : Seq[String], keys2 : Seq[String], frame2 : DynamicFrame, transformationContext : String = "", callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` + `keys1` – 此 `DynamicFrame` 中用于联接的列。 + `keys2` – `frame2` 中用于联接的列。必须与 `keys1` 的长度相同。 + `frame2` – 要联接的 `DynamicFrame`。返回使用指定的键对 `frame2` 执行 equijoin 的结果。 ## Def map ``` def map( f : DynamicRecord => DynamicRecord, errorMsg : String = "", transformationContext : String = "", callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` 返回通过对此 `DynamicFrame` 中的每个记录应用指定函数“`f`”构造的新 `DynamicFrame`。此方法先复制每个记录，然后再应用指定函数，因此可以安全地转变记录。如果映射函数在给定记录上引发异常，则该记录将标记为错误，并且堆栈跟踪将另存为错误记录中的一个列。 ## Def mergeDynamicFrames ``` def mergeDynamicFrames( stageDynamicFrame: DynamicFrame, primaryKeys: Seq[String], transformationContext: String = "", options: JsonOptions = JsonOptions.empty, callSite: CallSite = CallSite("Not provided"), stageThreshold: Long = 0, totalThreshold: Long = 0): DynamicFrame ``` + `stageDynamicFrame` – 要合并的暂存 `DynamicFrame`。 + `primaryKeys` – 要匹配源和暂存 `DynamicFrame` 中的记录的主键字段列表。 + `transformationContext` – 用于检索有关当前转换的元数据的唯一字符串（可选）。 + `options` – 为此转换提供其他信息的 JSON 名称-值对的字符串。 + `callSite` – 用于为错误报告提供上下文信息。 + `stageThreshold` — 一个 `Long`。给定转换中处理需要排除的错误的数目。 + `totalThreshold` — 一个 `Long`。此转换中处理需要排除的错误的总数。基于指定主键的将此 `DynamicFrame` 与暂存 `DynamicFrame` 合并以标识记录。不会对重复记录（具有相同主键的记录）去除重复。如果暂存帧中没有匹配的记录，则从源中保留所有记录（包括重复记录）。如果暂存帧具有匹配的记录，则暂存帧中的记录将覆盖 AWS Glue 中的源中的记录。在以下情况下，返回的 `DynamicFrame` 将包含记录 A： 1. 如果 `A` 在源帧和暂存帧中都存在，则返回暂存帧中的 `A`。 1. 如果 `A` 在源表中，且`A.primaryKeys` 不在 `stagingDynamicFrame` 中（这意味着，未在暂存表中更新 `A`）。源帧和暂存帧不需要具有相同的架构。 **Example** ``` val mergedFrame: DynamicFrame = srcFrame.mergeDynamicFrames(stageFrame, Seq("id1", "id2")) ``` ## Def printSchema ``` def printSchema : Unit ``` 以人类可读的格式将此 `DynamicFrame` 的架构输出到 `stdout`。 ## Def recomputeSchema ``` def recomputeSchema : Schema ``` 强制重新计算架构。这需要扫描数据，但如果数据中不存在当前架构中的一些字段，则可能会“压缩”架构。返回经过重新计算的架构。 ## Def relationalize ``` def relationalize( rootTableName : String, stagingPath : String, options : JsonOptions = JsonOptions.empty, transformationContext : String = "", callSite : CallSite = CallSite("Not provided"), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : Seq[DynamicFrame] ``` + `rootTableName` – 在输出中用于基本 `DynamicFrame` 的名称。通过透视数组创建的 `DynamicFrame` 以此作为前缀开头。 + `stagingPath` – 用于写入中间数据的 Amazon Simple Storage Service（Amazon S3）路径。 + `options` – 关系化选项和配置。目前未使用。展平所有嵌套的结构并将数组透视为单独的表。您可以使用此操作准备深度嵌套的数据以提取到关系数据库中。使用与 [Unnest](#glue-etl-scala-apis-glue-dynamicframe-class-defs-unnest) 转换相同的方式展平嵌套结构。此外，将数组透视为单独的表，其中每个数组元素成为一行。例如，假设您有一个包含以下数据的 `DynamicFrame`。 ``` {"name": "Nancy", "age": 47, "friends": ["Fred", "Lakshmi"]} {"name": "Stephanie", "age": 28, "friends": ["Yao", "Phil", "Alvin"]} {"name": "Nathan", "age": 54, "friends": ["Nicolai", "Karen"]} ``` 运行以下代码。 ``` {{{ df.relationalize("people", "s3:/my_bucket/my_path", JsonOptions.empty) }}} ``` 这会生成两个表。第一个表名为“people”，其中包含以下内容。 ``` {{{ {"name": "Nancy", "age": 47, "friends": 1} {"name": "Stephanie", "age": 28, "friends": 2} {"name": "Nathan", "age": 54, "friends": 3) }}} ``` 在这里，friends 数组已被替换为自动生成的联接键。创建了名为 `people.friends` 的单独表，其中包含以下内容。 ``` {{{ {"id": 1, "index": 0, "val": "Fred"} {"id": 1, "index": 1, "val": "Lakshmi"} {"id": 2, "index": 0, "val": "Yao"} {"id": 2, "index": 1, "val": "Phil"} {"id": 2, "index": 2, "val": "Alvin"} {"id": 3, "index": 0, "val": "Nicolai"} {"id": 3, "index": 1, "val": "Karen"} }}} ``` 在此表中，“`id`”是标识数组元素来自哪个记录的联接键，“`index`”是指在原始数组中的位置，“`val`”是实际的数组条目。 `relationalize` 方法返回通过以递归方式对所有数组应用此过程创建的 `DynamicFrame` 序列。 **注意** AWS Glue 库自动为新表生成联接键。为确保联接键在作业运行中是唯一的，您必须启用作业书签。 ## Def renameField ``` def renameField( oldName : String, newName : String, transformationContext : String = "", callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` + `oldName` – 列的原始名称。 + `newName` – 列的新名称。返回新 `DynamicFrame`，其中的指定字段进行了重命名。您可以使用此方法重命名嵌套字段。例如，以下代码在 address 结构中将 `state` 重命名为 `state_code`。 ``` {{{ df.renameField("address.state", "address.state_code") }}} ``` ## Def repartition ``` def repartition( numPartitions : Int, transformationContext : String = "", callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` 返回具有 `numPartitions` 分区的新 `DynamicFrame`。 ## Def resolveChoice ``` def resolveChoice( specs : Seq[Product2[String, String]] = Seq.empty[ResolveSpec], choiceOption : Option[ChoiceOption] = None, database : Option[String] = None, tableName : Option[String] = None, transformationContext : String = "", callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` + `choiceOption` – 应用于 specs 序列中未列出的所有 `ChoiceType` 列的操作。 + `database` – 与 `match_catalog` 操作一起使用的数据目录数据库。 + `tableName` – 与 `match_catalog` 操作一起使用的数据目录表。通过将一个或多个 `ChoiceType` 替换为更具体的类型来返回新 `DynamicFrame`。可通过两种方式使用 `resolveChoice`。第一种是指定一系列特定列以及如何解析它们。这些指定为由 (列, 操作) 对组成的元组。可能的操作如下： + `cast:type` – 尝试将所有值转换为指定的类型。 + `make_cols` – 将每个不同的类型转换为名为 `columnName_type` 的列。 + `make_struct` – 将列转换为具有每个不同类型的键的结构。 + `project:type` – 仅保留指定类型的值。 `resolveChoice` 的另一种模式是为所有 `ChoiceType` 指定单个解析方法。这可以在执行前不知道 `ChoiceType` 的完整列表的情况下使用。除了上面列出的操作外，此模式还支持以下操作： + `match_catalog``ChoiceType` – 尝试将每个转换为指定目录表中的对应类型。 **示例：** 通过转换为 int 来解析 `user.id` 列，并使 `address` 字段仅保留结构。 ``` {{{ df.resolveChoice(specs = Seq(("user.id", "cast:int"), ("address", "project:struct"))) }}} ``` 通过将每个选择转换为单独的列来解析所有 `ChoiceType`。 ``` {{{ df.resolveChoice(choiceOption = Some(ChoiceOption("make_cols"))) }}} ``` 通过转换为指定目录表中的类型来解析所有 `ChoiceType`。 ``` {{{ df.resolveChoice(choiceOption = Some(ChoiceOption("match_catalog")), database = Some("my_database"), tableName = Some("my_table")) }}} ``` ## Def schema ``` def schema : Schema ``` 返回此 `DynamicFrame` 的架构。保证返回的架构包含此 `DynamicFrame` 中的记录中存在的每个字段。但在少数情况下，它可能还包含其他字段。您可以使用 [Unnest](#glue-etl-scala-apis-glue-dynamicframe-class-defs-unnest) 方法基于此 `DynamicFrame` 中的记录来“压缩”架构。 ## Def selectField ``` def selectField( fieldName : String, transformationContext : String = "", callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` 返回作为 `DynamicFrame` 的单个字段。 ## Def selectFields ``` def selectFields( paths : Seq[String], transformationContext : String = "", callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` + `paths` – 要选择的列名称序列。返回包含指定列的新 `DynamicFrame`。 **注意** 您只能使用 `selectFields` 方法选择顶级列。您可以使用 [applyMapping](#glue-etl-scala-apis-glue-dynamicframe-class-defs-applyMapping) 方法选择嵌套列。 ## Def show ``` def show( numRows : Int = 20 ) : Unit ``` + `numRows` – 要输出的行数。以 JSON 格式输出此 `DynamicFrame` 中的行。 ## Def simplifyDDBJson 使用 AWS Glue DynamoDB 导出连接器进行 DynamoDB 导出时会生成具有特定嵌套结构的 JSON 文件。有关更多信息，请参阅[数据对象](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/S3DataExport.Output.html)。`simplifyDDBJson`简化此数据类型的 DynamicFrame 中的嵌套列，并返回新的简化 DynamicFrame。如果 List 类型中包含多种类型或 Map 类型，则不会简化 List 中的元素。此方法仅支持 DynamoDB 导出 JSON 格式的数据。考虑使用 `unnest` 来对其他类型的数据进行类似更改。 ``` def simplifyDDBJson() : DynamicFrame ``` 此方法不使用任何参数。 **示例输入** 考虑 DynamoDB 导出生成的以下架构： ``` root |-- Item: struct | |-- parentMap: struct | | |-- M: struct | | | |-- childMap: struct | | | | |-- M: struct | | | | | |-- appName: struct | | | | | | |-- S: string | | | | | |-- packageName: struct | | | | | | |-- S: string | | | | | |-- updatedAt: struct | | | | | | |-- N: string | |-- strings: struct | | |-- SS: array | | | |-- element: string | |-- numbers: struct | | |-- NS: array | | | |-- element: string | |-- binaries: struct | | |-- BS: array | | | |-- element: string | |-- isDDBJson: struct | | |-- BOOL: boolean | |-- nullValue: struct | | |-- NULL: boolean ``` **示例代码** ``` import com.amazonaws.services.glue.GlueContext import com.amazonaws.services.glue.util.GlueArgParser import com.amazonaws.services.glue.util.Job import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.DynamoDbDataSink import org.apache.spark.SparkContextimport scala.collection.JavaConverters._ object GlueApp { def main(sysArgs: Array[String]): Unit = { val glueContext = new GlueContext(SparkContext.getOrCreate()) val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray) Job.init(args("JOB_NAME"), glueContext, args.asJava) val dynamicFrame = glueContext.getSourceWithFormat( connectionType = "dynamodb", options = JsonOptions(Map( "dynamodb.export" -> "ddb", "dynamodb.tableArn" -> "ddbTableARN", "dynamodb.s3.bucket" -> "exportBucketLocation", "dynamodb.s3.prefix" -> "exportBucketPrefix", "dynamodb.s3.bucketOwner" -> "exportBucketAccountID", )) ).getDynamicFrame() val simplified = dynamicFrame.simplifyDDBJson() simplified.printSchema() Job.commit() } } ``` ### 输出示例 `simplifyDDBJson` 转换会将其简化为： ``` root |-- parentMap: struct | |-- childMap: struct | | |-- appName: string | | |-- packageName: string | | |-- updatedAt: string |-- strings: array | |-- element: string |-- numbers: array | |-- element: string |-- binaries: array | |-- element: string |-- isDDBJson: boolean |-- nullValue: null ``` ## Def spigot ``` def spigot( path : String, options : JsonOptions = new JsonOptions("{}"), transformationContext : String = "", callSite : CallSite = CallSite("Not provided"), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` 返回相同记录但写出一部分记录作为副作用的传递转换。 + `path` – 将输出写入的 Amazon S3 中的路径，格式为 `s3://bucket//path`。 + `options` – 描述取样行为的可选的 `JsonOptions` 映射。返回包含与这一个相同的记录的 `DynamicFrame`。默认情况下，将 100 个任意记录写入通过 `path` 指定的位置。您可以使用 `options` 映射自定义此行为。有效键包括： + `topk` – 指定写出的记录的总数。默认值为 100。 + `prob` – 指定包含单个记录的概率（小数形式）。默认值为 1。例如，以下调用将对数据集进行取样，方法是以 20% 的概率选择每个记录并在写入 200 个记录后停止。 ``` {{{ df.spigot("s3://my_bucket/my_path", JsonOptions(Map("topk" -> 200, "prob" -> 0.2))) }}} ``` ## Def splitFields ``` def splitFields( paths : Seq[String], transformationContext : String = "", callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : Seq[DynamicFrame] ``` + `paths` — 包括在第一个 `DynamicFrame` 中的路径。返回两个 `DynamicFrame` 的序列。第一个 `DynamicFrame` 包含指定路径，第二个包含所有其他列。 **示例** 此示例采用从 AWS Glue 数据目录中的 `legislators` 数据库中的 `persons` 表创建的 DynamicFrame，并将 DynamicFrame 拆分为两个，其中指定的字段进入第一个 DynamicFrame，其余字段进入第二个 DynamicFrame。然后，示例将从结果中选择第一个 DynamicFrame。 ``` val InputFrame = glueContext.getCatalogSource(database="legislators", tableName="persons", transformationContext="InputFrame").getDynamicFrame() val SplitField_collection = InputFrame.splitFields(paths=Seq("family_name", "name", "links.note", "links.url", "gender", "image", "identifiers.scheme", "identifiers.identifier", "other_names.lang", "other_names.note", "other_names.name"), transformationContext="SplitField_collection") val ResultFrame = SplitField_collection(0) ``` ## Def splitRows ``` def splitRows( paths : Seq[String], values : Seq[Any], operators : Seq[String], transformationContext : String, callSite : CallSite, stageThreshold : Long, totalThreshold : Long ) : Seq[DynamicFrame] ``` 基于将列与常量比较的谓词来拆分行。 + `paths` – 用于比较的列。 + `values` – 用于比较的常量值。 + `operators` – 用于比较的运算符。返回两个 `DynamicFrame` 的序列。第一个包含其谓词为 true 的行，第二个包含其谓词为 false 的行。使用三个序列指定谓词：“`paths`”包含（可能嵌套的）列名称，“`values`”包含要与其进行比较的常量值，“`operators`”包含用于比较的运算符。所有这三个序列必须长度相同：第 `n` 个运算符将用于将第 `n` 个列与第 `n` 个值进行比较。每个运算符必须是以下运算符之一：“`!=`”、“`=`”、“`<=`”、“`<`”、“`>=`”或“`>`”。例如，以下调用将拆分 `DynamicFrame`，以便第一个输出帧将包含美国 65 岁以上的人员记录，第二个将包含所有其他记录。 ``` {{{ df.splitRows(Seq("age", "address.country"), Seq(65, "USA"), Seq(">=", "=")) }}} ``` ## Def stageErrorsCount ``` def stageErrorsCount ``` 返回在计算此 `DynamicFrame` 时创建的错误记录数。这不包括作为输入传递给此 `DynamicFrame` 的以前操作中的错误。 ## Def toDF ``` def toDF( specs : Seq[ResolveSpec] = Seq.empty[ResolveSpec] ) : DataFrame ``` 将此 `DynamicFrame` 转换为具有相同架构和记录的 Apache Spark SQL `DataFrame`。 **注意** 由于 `DataFrame` 不支持 `ChoiceType`，因此此方法自动将 `ChoiceType` 列转换为 `StructType`。有关更多信息和用于解析选择的选项，请参阅 [resolveChoice](#glue-etl-scala-apis-glue-dynamicframe-class-defs-resolveChoice)。 ## Def unbox ``` def unbox( path : String, format : String, optionString : String = "{}", transformationContext : String = "", callSite : CallSite = CallSite("Not provided"), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` + `path` – 要分析的列。必须是字符串或二进制。 + `format` – 用于分析的格式。 + `optionString` – 传递到格式的选项，如 CSV 分隔符。根据指定的格式分析嵌入式字符串或二进制列。已分析的列嵌套在具有原始列名称的结构下。例如，假设您具有包含一个嵌入式 JSON 列的 CSV 文件。 ``` name, age, address Sally, 36, {"state": "NE", "city": "Omaha"} ... ``` 在初始分析后，您将获取具有以下架构的 `DynamicFrame`。 ``` {{{ root |-- name: string |-- age: int |-- address: string }}} ``` 您可以对地址列调用 `unbox` 来分析特定组件。 ``` {{{ df.unbox("address", "json") }}} ``` 这为我们提供了具有以下架构的 `DynamicFrame`。 ``` {{{ root |-- name: string |-- age: int |-- address: struct | |-- state: string | |-- city: string }}} ``` ## Def unnest ``` def unnest( transformationContext : String = "", callSite : CallSite = CallSite("Not Provided"), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` 返回平展了所有嵌套结构的新 `DynamicFrame`。构造名称时使用“`.`”（句点）字符。例如，假设您有一个包含以下架构的 `DynamicFrame`。 ``` {{{ root |-- name: string |-- age: int |-- address: struct | |-- state: string | |-- city: string }}} ``` 以下调用取消嵌套 address 结构。 ``` {{{ df.unnest() }}} ``` 生成的架构如下所示。 ``` {{{ root |-- name: string |-- age: int |-- address.state: string |-- address.city: string }}} ``` 此方法还取消嵌套数组中的嵌套结构。但由于历史原因，此类字段的名称前附加了括起来的数组的名称和“`.val`”。 ## Def unnestDDBJson ``` unnestDDBJson(transformationContext : String = "", callSite : CallSite = CallSite("Not Provided"), stageThreshold : Long = 0, totalThreshold : Long = 0): DynamicFrame ``` 解除 `DynamicFrame` 中的嵌套列，具体位于 DynamoDB JSON 结构中，并返回一个新的非嵌套 `DynamicFrame`。属于结构类型数组的列将不会被解除嵌套。请注意，这是一种特定类型的非嵌套转换，其行为与常规 `unnest` 转换不同，并且要求数据已存在于 DynamoDB JSON 结构中。有关更多信息，请参阅 [DynamoDB JSON](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/DataExport.Output.html#DataExport.Output.Data)。例如，使用 DynamoDB JSON 结构读取导出的架构可能如下所示： ``` root |-- Item: struct | |-- ColA: struct | | |-- S: string | |-- ColB: struct | | |-- S: string | |-- ColC: struct | | |-- N: string | |-- ColD: struct | | |-- L: array | | | |-- element: null ``` `unnestDDBJson()` 转换会将此转换为： ``` root |-- ColA: string |-- ColB: string |-- ColC: string |-- ColD: array | |-- element: null ``` 以下代码示例演示了如何使用 AWS Glue DynamoDB 导出连接器、调用 DynamoDB JSON 解除嵌套命令，以及打印分区数量： ``` import com.amazonaws.services.glue.GlueContext import com.amazonaws.services.glue.util.GlueArgParser import com.amazonaws.services.glue.util.Job import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.DynamoDbDataSink import org.apache.spark.SparkContext import scala.collection.JavaConverters._ object GlueApp { def main(sysArgs: Array[String]): Unit = { val glueContext = new GlueContext(SparkContext.getOrCreate()) val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray) Job.init(args("JOB_NAME"), glueContext, args.asJava) val dynamicFrame = glueContext.getSourceWithFormat( connectionType = "dynamodb", options = JsonOptions(Map( "dynamodb.export" -> "ddb", "dynamodb.tableArn" -> "", "dynamodb.s3.bucket" -> "", "dynamodb.s3.prefix" -> "", "dynamodb.s3.bucketOwner" -> "", )) ).getDynamicFrame() val unnested = dynamicFrame.unnestDDBJson() print(unnested.getNumPartitions()) Job.commit() } } ``` ## Def withFrameSchema ``` def withFrameSchema( getSchema : () => Schema ) : DynamicFrame ``` + `getSchema` – 返回要使用的架构的函数。指定为零参数函数来推迟可能昂贵的计算。将此 `DynamicFrame` 的架构设置为指定值。这主要在内部使用以避免成本高昂的架构重新计算。传入的架构必须包含数据中存在的所有列。 ## Def withName ``` def withName( name : String ) : DynamicFrame ``` + `name` – 要使用的新名称。返回使用新名称的此 `DynamicFrame` 的副本。 ## Def withTransformationContext ``` def withTransformationContext( ctx : String ) : DynamicFrame ``` 返回具有指定转换上下文的此 `DynamicFrame` 的副本。 # DynamicFrame 对象 **程序包：com.amazonaws.services.glue** ``` object DynamicFrame ``` ## Def apply ``` def apply( df : DataFrame, glueContext : GlueContext ) : DynamicFrame ``` ## Def emptyDynamicFrame ``` def emptyDynamicFrame( glueContext : GlueContext ) : DynamicFrame ``` ## Def fromPythonRDD ``` def fromPythonRDD( rdd : JavaRDD[Array[Byte]], glueContext : GlueContext ) : DynamicFrame ``` ## Def ignoreErrors ``` def ignoreErrors( fn : DynamicRecord => DynamicRecord ) : DynamicRecord ``` ## Def inlineErrors ``` def inlineErrors( msg : String, callSite : CallSite ) : (DynamicRecord => DynamicRecord) ``` ## Def newFrameWithErrors ``` def newFrameWithErrors( prevFrame : DynamicFrame, rdd : RDD[DynamicRecord], name : String = "", transformationContext : String = "", callSite : CallSite, stageThreshold : Long, totalThreshold : Long ) : DynamicFrame ``` # AWS Glue Scala DynamicRecord 类 **Topics** + [Def addField](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-addField) + [Def dropField](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-dropField) + [Def setError](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-setError) + [Def isError](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-isError) + [Def getError](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-getError) + [Def clearError](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-clearError) + [Def write](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-write) + [Def readFields](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-readFields) + [Def clone](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-clone) + [Def schema](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-schema) + [Def getRoot](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-getRoot) + [Def toJson](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-toJson) + [Def getFieldNode](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-getFieldNode) + [Def getField](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-getField) + [Def hashCode](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-hashCode) + [Def equals](#glue-etl-scala-apis-glue-dynamicrecord-class-defs-equals) + [DynamicRecord 对象](#glue-etl-scala-apis-glue-dynamicrecord-object) + [RecordTraverser 特性](#glue-etl-scala-apis-glue-recordtraverser-trait) **程序包：com.amazonaws.services.glue** ``` class DynamicRecord extends Serializable with Writable with Cloneable ``` `DynamicRecord` 是一个自描述数据结构，它表示被处理的数据集中的一行数据。自描述的意义在于，您可以通过检查记录本身来获取 `DynamicRecord` 所表示行的架构。`DynamicRecord` 类似于 Apache Spark 中的 `Row`。 ## Def addField ``` def addField( path : String, dynamicNode : DynamicNode ) : Unit ``` 将 [DynamicNode](glue-etl-scala-apis-glue-types-dynamicnode.md) 添加至指定路径。 + `path` – 要添加的字段的路径。 + `dynamicNode` – 要在指定路径处添加的 [DynamicNode](glue-etl-scala-apis-glue-types-dynamicnode.md)。 ## Def dropField ``` def dropField(path: String, underRename: Boolean = false): Option[DynamicNode] ``` 从指定路径删除 [DynamicNode](glue-etl-scala-apis-glue-types-dynamicnode.md) 并在指定路径中不存在阵列时返回已删除的节点。 + `path` – 要删除的字段的路径。 + `underRename``dropField` – 如果作为重命名转换的一部分调用，则为 True，否则为 false (默认为 false)。返回 `scala.Option Option` ([DynamicNode](glue-etl-scala-apis-glue-types-dynamicnode.md))。 ## Def setError ``` def setError( error : Error ) ``` 按 `error` 参数指定，将此记录设置为错误记录。返回 `DynamicRecord`。 ## Def isError ``` def isError ``` 检查此记录是否为错误记录。 ## Def getError ``` def getError ``` 如果记录为错误记录，则获取 `Error`。如果此记录为错误记录，则返回 `scala.Some Some` (错误)；否则返回 `scala.None`。 ## Def clearError ``` def clearError ``` 将 `Error` 设置为 `scala.None.None`。 ## Def write ``` override def write( out : DataOutput ) : Unit ``` ## Def readFields ``` override def readFields( in : DataInput ) : Unit ``` ## Def clone ``` override def clone : DynamicRecord ``` 将此记录克隆到新 `DynamicRecord` 并将其返回。 ## Def schema ``` def schema ``` 通过检查记录获取 `Schema`。 ## Def getRoot ``` def getRoot : ObjectNode ``` 获取记录的根 `ObjectNode`。 ## Def toJson ``` def toJson : String ``` 获取记录的 JSON 字符串。 ## Def getFieldNode ``` def getFieldNode( path : String ) : Option[DynamicNode] ``` 获取指定 `path` 处的字段值，作为 `DynamicNode` 的 Option。如果字段存在，则返回 `scala.Some Some` ([DynamicNode](glue-etl-scala-apis-glue-types-dynamicnode.md))，否则返回 `scala.None.None`。 ## Def getField ``` def getField( path : String ) : Option[Any] ``` 获取指定 `path` 处的字段值，作为 `DynamicNode` 的 Option。返回 `scala.Some Some` (值)。 ## Def hashCode ``` override def hashCode : Int ``` ## Def equals ``` override def equals( other : Any ) ``` ## DynamicRecord 对象 ``` object DynamicRecord ``` ### Def apply ``` def apply( row : Row, schema : SparkStructType ) ``` 应用方法以将 Apache Spark SQL `Row` 转换为 [DynamicRecord](#glue-etl-scala-apis-glue-dynamicrecord-class)。 + `row` – Spark SQL `Row`。 + `schema` – 该行的 `Schema`。返回 `DynamicRecord`。 ## RecordTraverser 特性 ``` trait RecordTraverser { def nullValue(): Unit def byteValue(value: Byte): Unit def binaryValue(value: Array[Byte]): Unit def booleanValue(value: Boolean): Unit def shortValue(value: Short) : Unit def intValue(value: Int) : Unit def longValue(value: Long) : Unit def floatValue(value: Float): Unit def doubleValue(value: Double): Unit def decimalValue(value: BigDecimal): Unit def stringValue(value: String): Unit def dateValue(value: Date): Unit def timestampValue(value: Timestamp): Unit def objectStart(length: Int): Unit def objectKey(key: String): Unit def objectEnd(): Unit def mapStart(length: Int): Unit def mapKey(key: String): Unit def mapEnd(): Unit def arrayStart(length: Int): Unit def arrayEnd(): Unit } ``` # AWS Glue Scala GlueContext API **程序包：com.amazonaws.services.glue** ``` class GlueContext extends SQLContext(sc) ( @transient val sc : SparkContext, val defaultSourcePartitioner : PartitioningStrategy ) ``` `GlueContext` 是在以下位置读取和写入 [DynamicFrame](glue-etl-scala-apis-glue-dynamicframe.md) 的入口点：Amazon Simple Storage Service（Amazon S3）、AWS Glue 数据目录、JDBC 等。此类提供实用程序函数，用于创建可用来读取和写入 `DynamicFrame` 的 [数据源特性](glue-etl-scala-apis-glue-datasource-trait.md) 和 [DataSink](glue-etl-scala-apis-glue-datasink-class.md) 对象。如果从源创建的分区数小于分区的最小阈值（默认值为 10），您还可以使用 `GlueContext` 设置 `DynamicFrame` 中的目标分区数量（默认值为 20）。 ## def addIngestionTimeColumns ``` def addIngestionTimeColumns( df : DataFrame, timeGranularity : String = "") : dataFrame ``` 将提取时间列（例如 `ingest_year`、`ingest_month`、`ingest_day`、`ingest_hour`、`ingest_minute`）附加到输入 `DataFrame`。当您指定以 Amazon S3 为目标的数据目录表时，AWS Glue 生成的脚本中会自动生成此函数。此函数使用输出表上的提取时间列自动更新分区。这允许根据提取时间自动对输出数据进行分区，而不需要在输入数据中显示提取时间列。 + `dataFrame` – 提取时间列要附加到的 `dataFrame`。 + `timeGranularity` – 时间列的粒度。有效值为“`day`”、“`hour`”和“`minute`”。例如，如果“`hour`”传递到函数，原始 `dataFrame` 将附加“`ingest_year`”、“`ingest_month`”、“`ingest_day`”和“`ingest_hour`”时间列。在附加时间粒度列后返回数据框。示例： ``` glueContext.addIngestionTimeColumns(dataFrame, "hour") ``` ## def createDataFrameFromOptions ``` def createDataFrameFromOptions( connectionType : String, connectionOptions : JsonOptions, transformationContext : String = "", format : String = null, formatOptions : JsonOptions = JsonOptions.empty ) : DataSource ``` 返回一个使用指定连接和格式创建的 `DataFrame`。仅将此函数与 AWS Glue 流式传输源结合使用。 + `connectionType` – 流式传输连接类型。有效值包括 `kinesis` 和 `kafka`。 + `connectionOptions` – 连接选项，不同于 Kinesis 和 Kafka。您可以在 [AWS Glue for Spark 中适用于 ETL 的连接类型和选项](aws-glue-programming-etl-connect.md) 找到每个流式传输数据源的所有连接选项列表。请注意流式传输连接选项的以下差异： + Kinesis 流式传输源需要 `streamARN`、`startingPosition`、`inferSchema` 和 `classification`。 + Kinesis 流式传输源需要 `connectionName`、`topicName`、`startingOffsets`、`inferSchema` 和 `classification`。 + `transformationContext` – 要使用的转换上下文（可选）。 + `format` – 格式规范（可选）。这用于 Amazon S3 或支持多种格式的 AWS Glue 连接。有关所支持格式的信息，请参阅 [AWS Glue for Spark 中的输入和输出的数据格式选项](aws-glue-programming-etl-format.md)。 + `formatOptions` – 指定格式的格式选项。有关所支持的格式选项的信息，请参阅[数据格式选项](aws-glue-programming-etl-format.md)。 Amazon Kinesis 流式传输源示例： ``` val data_frame_datasource0 = glueContext.createDataFrameFromOptions(transformationContext = "datasource0", connectionType = "kinesis", connectionOptions = JsonOptions("""{"streamName": "example_stream", "startingPosition": "TRIM_HORIZON", "inferSchema": "true", "classification": "json"}}""")) ``` Kafka 流式传输源示例： ``` val data_frame_datasource0 = glueContext.createDataFrameFromOptions(transformationContext = "datasource0", connectionType = "kafka", connectionOptions = JsonOptions("""{"connectionName": "example_connection", "topicName": "example_topic", "startingPosition": "earliest", "inferSchema": "false", "classification": "json", "schema":"`column1` STRING, `column2` STRING"}""")) ``` ## forEachBatch **`forEachBatch(frame, batch_function, options)`** 将传入的 `batch_function` 应用于从流式传输源读取的每个微批处理。 + `frame` – 包含当前微处理的 DataFrame。 + `batch_function` – 应用于每个微处理的函数。 + `options` – 键值对集合，其中包含有关如何处理微批处理的信息。以下选项为必填： + `windowSize` – 处理每个批处理所花费的时间量。 + `checkpointLocation` – 为流式传输 ETL 任务存储检查点的位置。 + `batchMaxRetries` – 在该批处理失败时重试的最大次数。默认值为 3。此选项仅适用于 Glue 版本 2.0 及更高版本。 **示例：** ``` glueContext.forEachBatch(data_frame_datasource0, (dataFrame: Dataset[Row], batchId: Long) => { if (dataFrame.count() > 0) { val datasource0 = DynamicFrame(glueContext.addIngestionTimeColumns(dataFrame, "hour"), glueContext) // @type: DataSink // @args: [database = "tempdb", table_name = "fromoptionsoutput", stream_batch_time = "100 seconds", // stream_checkpoint_location = "s3://from-options-testing-eu-central-1/fromOptionsOutput/checkpoint/", // transformation_ctx = "datasink1"] // @return: datasink1 // @inputs: [frame = datasource0] val options_datasink1 = JsonOptions( Map("partitionKeys" -> Seq("ingest_year", "ingest_month","ingest_day", "ingest_hour"), "enableUpdateCatalog" -> true)) val datasink1 = glueContext.getCatalogSink( database = "tempdb", tableName = "fromoptionsoutput", redshiftTmpDir = "", transformationContext = "datasink1", additionalOptions = options_datasink1).writeDynamicFrame(datasource0) } }, JsonOptions("""{"windowSize" : "100 seconds", "checkpointLocation" : "s3://from-options-testing-eu-central-1/fromOptionsOutput/checkpoint/"}""")) ``` ## def getCatalogSink ``` def getCatalogSink( database : String, tableName : String, redshiftTmpDir : String = "", transformationContext : String = "" additionalOptions: JsonOptions = JsonOptions.empty, catalogId: String = null ) : DataSink ``` 创建一个可向在数据目录中定义的表中指定的位置写入的 [DataSink](glue-etl-scala-apis-glue-datasink-class.md)。 + `database` – 数据目录中的数据库名称。 + `tableName` – 数据目录中的表名称。 + `redshiftTmpDir` – 用于某些数据接收器的临时暂存目录。设置为默认情况下为空。 + `transformationContext` – 与任务书签要使用的接收器关联的转换上下文。设置为默认情况下为空。 + `additionalOptions` 提供给 AWS Glue 的额外选项。 + `catalogId` – 正在访问的数据目录 ID（账户 ID）。当为空时，将使用调用方的默认账户 ID。返回 `DataSink`。 ## def getCatalogSource ``` def getCatalogSource( database : String, tableName : String, redshiftTmpDir : String = "", transformationContext : String = "" pushDownPredicate : String = " " additionalOptions: JsonOptions = JsonOptions.empty, catalogId: String = null ) : DataSource ``` 创建可从数据目录中的表定义读取数据的 [数据源特性](glue-etl-scala-apis-glue-datasource-trait.md)。 + `database` – 数据目录中的数据库名称。 + `tableName` – 数据目录中的表名称。 + `redshiftTmpDir` – 用于某些数据接收器的临时暂存目录。设置为默认情况下为空。 + `transformationContext` – 与任务书签要使用的接收器关联的转换上下文。设置为默认情况下为空。 + `pushDownPredicate` – 筛选分区，而不必列出并读取数据集中的所有文件。有关更多信息，请参阅 [使用下推谓词进行预筛选](aws-glue-programming-etl-partitions.md#aws-glue-programming-etl-partitions-pushdowns)。 + `additionalOptions` – 可选名称/值对的集合。可能选项包括 [AWS Glue for Spark 中适用于 ETL 的连接类型和选项](aws-glue-programming-etl-connect.md) 中列出的选项，但 `endpointUrl`、`streamName`、`bootstrap.servers`、`security.protocol`、`topicName`、`classification` 和 `delimiter` 除外。另一个支持的选项是 `catalogPartitionPredicate`： `catalogPartitionPredicate` – 要传递目录表达式以根据索引列进行筛选。这样会将筛选下推到服务器端。有关更多信息，请参阅 [AWS Glue 分区数据](https://docs.aws.amazon.com/glue/latest/dg/partition-indexes.html)。请注意，`push_down_predicate` 和 `catalogPartitionPredicate` 使用不同的语法。前者使用 Spark SQL 标准语法，后者使用 JSQL 解析器。 + `catalogId` – 正在访问的数据目录 ID（账户 ID）。当为空时，将使用调用方的默认账户 ID。返回 `DataSource`。 **流式传输源示例** ``` val data_frame_datasource0 = glueContext.getCatalogSource( database = "tempdb", tableName = "test-stream-input", redshiftTmpDir = "", transformationContext = "datasource0", additionalOptions = JsonOptions("""{ "startingPosition": "TRIM_HORIZON", "inferSchema": "false"}""") ).getDataFrame() ``` ## def getJDBCSink ``` def getJDBCSink( catalogConnection : String, options : JsonOptions, redshiftTmpDir : String = "", transformationContext : String = "", catalogId: String = null ) : DataSink ``` 创建一个可向在数据目录中的 `Connection` 对象指定的 JDBC 数据库写入的 [DataSink](glue-etl-scala-apis-glue-datasink-class.md)。`Connection` 对象具有用于连接 JDBC 接收器的信息，包括 URL、用户名、密码、VPC、子网和安全组。 + `catalogConnection` – 数据目录中包含要写入的 JDBC URL 的连接名称。 + `options` – JSON 名称-值对的字符串，提供写入 JDBC 数据存储所需的附加信息。这包括： + *dbtable*（必填）- JDBC 表名称。对于在数据库中支持架构的 JDBC 数据存储，指定 `schema.table-name`。如果未提供架构，则使用默认的“public”架构。以下示例显示一个指向名为 `test` 的架构的选项参数和数据库 `test_db` 中一个名为 `test_table` 的表。 ``` options = JsonOptions("""{"dbtable": "test.test_table", "database": "test_db"}""") ``` + *database*（必填）- JDBC 数据库名称。 + 任何其他选项直接传递到 SparkSQL JDBC 写入器。有关更多信息，请参阅 [Spark 的 Redshift 数据源](https://github.com/databricks/spark-redshift)。 + `redshiftTmpDir` – 用于某些数据接收器的临时目录。设置为默认情况下为空。 + `transformationContext` – 与任务书签要使用的接收器关联的转换上下文。设置为默认情况下为空。 + `catalogId` – 正在访问的数据目录 ID（账户 ID）。当为空时，将使用调用方的默认账户 ID。示例代码： ``` getJDBCSink(catalogConnection = "my-connection-name", options = JsonOptions("""{"dbtable": "my-jdbc-table", "database": "my-jdbc-db"}"""), redshiftTmpDir = "", transformationContext = "datasink4") ``` 返回 `DataSink`。 ## def getSink ``` def getSink( connectionType : String, connectionOptions : JsonOptions, transformationContext : String = "" ) : DataSink ``` 创建 [DataSink](glue-etl-scala-apis-glue-datasink-class.md)，将数据写入 Amazon Simple Storage Service（Amazon S3）、JDBC、AWS Glue 数据目录、Apache Kafka 或 Amazon Kinesis 数据流等目标写入数据。 + `connectionType` - 连接的类型。请参阅[AWS Glue for Spark 中适用于 ETL 的连接类型和选项](aws-glue-programming-etl-connect.md)。 + `connectionOptions` – JSON 名称-值对的字符串，提供与数据接收器建立连接所需的附加信息。请参阅[AWS Glue for Spark 中适用于 ETL 的连接类型和选项](aws-glue-programming-etl-connect.md)。 + `transformationContext` – 与任务书签要使用的接收器关联的转换上下文。设置为默认情况下为空。返回 `DataSink`。 ## def getSinkWithFormat ``` def getSinkWithFormat( connectionType : String, options : JsonOptions, transformationContext : String = "", format : String = null, formatOptions : JsonOptions = JsonOptions.empty ) : DataSink ``` 创建 [DataSink](glue-etl-scala-apis-glue-datasink-class.md)，将数据写入 Amazon S3、JDBC、数据目录、Apache Kafka 或 Amazon Kinesis 数据流等目标写入数据。此外还将设置要写出到目标的数据格式。 + `connectionType` - 连接的类型。请参阅[AWS Glue for Spark 中适用于 ETL 的连接类型和选项](aws-glue-programming-etl-connect.md)。 + `options` – JSON 名称-值对的字符串，提供与数据接收器建立连接所需的附加信息。请参阅[AWS Glue for Spark 中适用于 ETL 的连接类型和选项](aws-glue-programming-etl-connect.md)。 + `transformationContext` – 与任务书签要使用的接收器关联的转换上下文。设置为默认情况下为空。 + `format` – 要写出到目标的数据的格式。 + `formatOptions` – JSON 名称-值对的字符串，提供用于设置目标处的数据格式的附加选项。请参阅[数据格式选项](aws-glue-programming-etl-format.md)。返回 `DataSink`。 ## def getSource ``` def getSource( connectionType : String, connectionOptions : JsonOptions, transformationContext : String = "" pushDownPredicate ) : DataSource ``` 创建可从 Amazon S3、JDBC 或 AWS Glue 数据目录等源中读取数据的 [数据源特性](glue-etl-scala-apis-glue-datasource-trait.md)。还支持 Kafka 和 Kinesis 流式传输数据源。 + `connectionType` – 数据源的类型。请参阅[AWS Glue for Spark 中适用于 ETL 的连接类型和选项](aws-glue-programming-etl-connect.md)。 + `connectionOptions` – JSON 名称-值对的字符串，提供与数据源建立连接所需的附加信息。有关更多信息，请参阅 [AWS Glue for Spark 中适用于 ETL 的连接类型和选项](aws-glue-programming-etl-connect.md)。 Kinesis 流式传输源需要以下连接选项：`streamARN`、`startingPosition`、`inferSchema` 和 `classification`。 Kinesis 流式传输源需要以下连接选项：`connectionName`、`topicName`、`startingOffsets`、`inferSchema` 和 `classification`。 + `transformationContext` – 与任务书签要使用的接收器关联的转换上下文。设置为默认情况下为空。 + `pushDownPredicate` – 预测分区列。返回 `DataSource`。 Amazon Kinesis 流式传输源示例： ``` val kinesisOptions = jsonOptions() data_frame_datasource0 = glueContext.getSource("kinesis", kinesisOptions).getDataFrame() private def jsonOptions(): JsonOptions = { new JsonOptions( s"""{"streamARN": "arn:aws:kinesis:eu-central-1:123456789012:stream/fromOptionsStream", |"startingPosition": "TRIM_HORIZON", |"inferSchema": "true", |"classification": "json"}""".stripMargin) } ``` Kafka 流式传输源示例： ``` val kafkaOptions = jsonOptions() val data_frame_datasource0 = glueContext.getSource("kafka", kafkaOptions).getDataFrame() private def jsonOptions(): JsonOptions = { new JsonOptions( s"""{"connectionName": "ConfluentKafka", |"topicName": "kafka-auth-topic", |"startingOffsets": "earliest", |"inferSchema": "true", |"classification": "json"}""".stripMargin) } ``` ## def getSourceWithFormat ``` def getSourceWithFormat( connectionType : String, options : JsonOptions, transformationContext : String = "", format : String = null, formatOptions : JsonOptions = JsonOptions.empty ) : DataSource ``` 创建一个 [数据源特性](glue-etl-scala-apis-glue-datasource-trait.md)，从 Amazon S3, JDBC 或 AWS Glue 数据目录等源读取数据并设置源中所存储数据的格式。 + `connectionType` – 数据源的类型。请参阅[AWS Glue for Spark 中适用于 ETL 的连接类型和选项](aws-glue-programming-etl-connect.md)。 + `options` – JSON 名称-值对的字符串，提供与数据源建立连接所需的附加信息。请参阅[AWS Glue for Spark 中适用于 ETL 的连接类型和选项](aws-glue-programming-etl-connect.md)。 + `transformationContext` – 与任务书签要使用的接收器关联的转换上下文。设置为默认情况下为空。 + `format` – 源中所存储数据的格式。当 `connectionType` 为“s3”时，您也可以指定 `format`。可以是以下值之一：“avro”、“csv”、“grokLog”、“ion”、“json”、“xml”、“parquet”或“orc”。 + `formatOptions` – JSON 名称-值对的字符串，提供用于在源中分析数据的附加选项。请参阅[数据格式选项](aws-glue-programming-etl-format.md)。返回 `DataSource`。 **示例** 从 Amazon S3 上逗号分隔值（CSV）文件的数据源创建 DynamicFrame： ``` val datasource0 = glueContext.getSourceWithFormat( connectionType="s3", options =JsonOptions(s"""{"paths": [ "s3://csv/nycflights.csv"]}"""), transformationContext = "datasource0", format = "csv", formatOptions=JsonOptions(s"""{"withHeader":"true","separator": ","}""") ).getDynamicFrame() ``` 使用 JDBC 连接从作为 PostgreSQL 的数据源创建 DynamicFrame： ``` val datasource0 = glueContext.getSourceWithFormat( connectionType="postgresql", options =JsonOptions(s"""{ "url":"jdbc:postgresql://databasePostgres-1.rds.amazonaws.com:5432/testdb", "dbtable": "public.company", "redshiftTmpDir":"", "user":"username", "password":"password123" }"""), transformationContext = "datasource0").getDynamicFrame() ``` 使用 JDBC 连接从作为 MySQL 的数据源创建 DynamicFrame： ``` val datasource0 = glueContext.getSourceWithFormat( connectionType="mysql", options =JsonOptions(s"""{ "url":"jdbc:mysql://databaseMysql-1.rds.amazonaws.com:3306/testdb", "dbtable": "athenatest_nycflights13_csv", "redshiftTmpDir":"", "user":"username", "password":"password123" }"""), transformationContext = "datasource0").getDynamicFrame() ``` ## def getSparkSession ``` def getSparkSession : SparkSession ``` 获取与此 GlueContext 关联的 `SparkSession` 对象。使用此 SparkSession 对象可注册要用于从 DynamicFrame 创建的 `DataFrame` 的表和 UDF。返回 SparkSession。 ## def startTransaction ``` def startTransaction(readOnly: Boolean):String ``` 开启新事务。内部调用 Lake Formation [startTransaction](https://docs.aws.amazon.com/lake-formation/latest/dg/aws-lake-formation-api-aws-lake-formation-api-transactions.html#aws-lake-formation-api-aws-lake-formation-api-transactions-StartTransaction) API。 + `readOnly` –（布尔值）指示此事务应为只读还是读写。使用只读事务 ID 进行的写入将被拒绝。只读事务不需要提交。返回事务 ID。 ## def commitTransaction ``` def commitTransaction(transactionId: String, waitForCommit: Boolean): Boolean ``` 尝试提交指定的事务。可能在事务完成提交之前返回 `commitTransaction`。内部调用 Lake Formation [commitTransaction](https://docs.aws.amazon.com/lake-formation/latest/dg/aws-lake-formation-api-aws-lake-formation-api-transactions.html#aws-lake-formation-api-aws-lake-formation-api-transactions-CommitTransaction) API。 + `transactionId` –（字符串）要提交的事务。 + `waitForCommit` –（布尔值）确定是否立即返回 `commitTransaction`。默认值为 true。如果为假，则轮询 `commitTransaction` 并等待事务提交。等待时间限制为 1 分钟使用指数回退，最多重试 6 次。返回布尔值，以指示是否完成提交。 ## def cancelTransaction ``` def cancelTransaction(transactionId: String): Unit ``` 尝试取消指定的事务。内部调用 Lake Formation [CancelTransaction](https://docs.aws.amazon.com/lake-formation/latest/dg/aws-lake-formation-api-aws-lake-formation-api-transactions.html#aws-lake-formation-api-aws-lake-formation-api-transactions-CancelTransaction) API。 + `transactionId` –（字符串）要取消的事务。如果事务以前已提交，则返回 `TransactionCommittedException` 异常。 ## def this ``` def this( sc : SparkContext, minPartitions : Int, targetPartitions : Int ) ``` 使用指定的 `SparkContext`、最小分区数和目标分区数创建 `GlueContext` 对象。 + `sc` — 这些区域有： `SparkContext`。 + `minPartitions` – 最小分区数。 + `targetPartitions` – 目标分区数。返回 `GlueContext`。 ## def this ``` def this( sc : SparkContext ) ``` 使用提供的 `GlueContext` 创建一个 `SparkContext` 对象。将最小分区数设置为 10，将目标分区数设置为 20。 + `sc` — 这些区域有： `SparkContext`。返回 `GlueContext`。 ## def this ``` def this( sparkContext : JavaSparkContext ) ``` 使用提供的 `GlueContext` 创建一个 `JavaSparkContext` 对象。将最小分区数设置为 10，将目标分区数设置为 20。 + `sparkContext` — 这些区域有： `JavaSparkContext`。返回 `GlueContext`。 # MappingSpec **程序包：com.amazonaws.services.glue** ## MappingSpec case 类 ``` case class MappingSpec( sourcePath: SchemaPath, sourceType: DataType, targetPath: SchemaPath, targetType: DataTyp ) extends Product4[String, String, String, String] { override def _1: String = sourcePath.toString override def _2: String = ExtendedTypeName.fromDataType(sourceType) override def _3: String = targetPath.toString override def _4: String = ExtendedTypeName.fromDataType(targetType) } ``` + `sourcePath` - 源字段的 `SchemaPath`。 + `sourceType` - 源字段的 `DataType`。 + `targetPath` - 目标字段的 `SchemaPath`。 + `targetType` - 目标字段的 `DataType`。 `MappingSpec` 指定从源路径和源数据类型到目标路径和目标数据类型的映射。源路径的源帧中的值会显示在目标路径的目标帧中。源数据类型将强制转换为目标数据类型。它从 `Product4` 扩展而来，因此您可在 `applyMapping` 接口中处理任何 `Product4`。 ## MappingSpec 对象 ``` object MappingSpec ``` `MappingSpec` 对象具有以下成员： ## Val orderingByTarget ``` val orderingByTarget: Ordering[MappingSpec] ``` ## Def apply ``` def apply( sourcePath : String, sourceType : DataType, targetPath : String, targetType : DataType ) : MappingSpec ``` 创建 `MappingSpec`。 + `sourcePath` – 源路径的字符串表示形式。 + `sourceType` - 源 `DataType`。 + `targetPath` – 目标路径的字符串表示形式。 + `targetType` - 目标 `DataType`。返回 `MappingSpec`。 ## Def apply ``` def apply( sourcePath : String, sourceTypeString : String, targetPath : String, targetTypeString : String ) : MappingSpec ``` 创建 `MappingSpec`。 + `sourcePath` – 源路径的字符串表示形式。 + `sourceType` – 源数据类型的字符串表示形式。 + `targetPath` – 目标路径的字符串表示形式。 + `targetType` – 目标数据类型的字符串表示形式。返回 MappingSpec。 ## Def apply ``` def apply( product : Product4[String, String, String, String] ) : MappingSpec ``` 创建 `MappingSpec`。 + `product` – 源路径、源数据类型、目标路径和目标数据类型的 `Product4`。返回 `MappingSpec`。 # AWS Glue Scala ResolveSpec API **Topics** + [ResolveSpec 对象](#glue-etl-scala-apis-glue-resolvespec-object) + [ResolveSpec case 类](#glue-etl-scala-apis-glue-resolvespec-case-class) **程序包：com.amazonaws.services.glue** ## ResolveSpec 对象 **ResolveSpec** ``` object ResolveSpec ``` ### Def ``` def apply( path : String, action : String ) : ResolveSpec ``` 创建 `ResolveSpec`。 + `path` – 需要解析的选择字段的字符串表示形式。 + `action` – 解析操作。操作可以是以下项之一：`Project`、`KeepAsStruct` 或 `Cast`。返回 `ResolveSpec`。 ### Def ``` def apply( product : Product2[String, String] ) : ResolveSpec ``` 创建 `ResolveSpec`。 + `product` - `Product2`：源路径、解析操作。返回 `ResolveSpec`。 ## ResolveSpec case 类 ``` case class ResolveSpec extends Product2[String, String] ( path : SchemaPath, action : String ) ``` 创建 `ResolveSpec`。 + `path` – 需要解析的选择字段的 `SchemaPath`。 + `action` – 解析操作。操作可以是以下项之一：`Project`、`KeepAsStruct` 或 `Cast`。 ### ResolveSpec def 方法 ``` def _1 : String ``` ``` def _2 : String ``` # AWS Glue Scala ArrayNode API **程序包：com.amazonaws.services.glue.types** ## ArrayNode case 类 **ArrayNode** ``` case class ArrayNode extends DynamicNode ( value : ArrayBuffer[DynamicNode] ) ``` ### ArrayNode def 方法 ``` def add( node : DynamicNode ) ``` ``` def clone ``` ``` def equals( other : Any ) ``` ``` def get( index : Int ) : Option[DynamicNode] ``` ``` def getValue ``` ``` def hashCode : Int ``` ``` def isEmpty : Boolean ``` ``` def nodeType ``` ``` def remove( index : Int ) ``` ``` def this ``` ``` def toIterator : Iterator[DynamicNode] ``` ``` def toJson : String ``` ``` def update( index : Int, node : DynamicNode ) ``` # AWS Glue Scala BinaryNode API **程序包：com.amazonaws.services.glue.types** ## BinaryNode case 类 **BinaryNode** ``` case class BinaryNode extends ScalarNode(value, TypeCode.BINARY) ( value : Array[Byte] ) ``` ### BinaryNode val 字段 + `ordering` ### BinaryNode def 方法 ``` def clone ``` ``` def equals( other : Any ) ``` ``` def hashCode : Int ``` # AWS Glue Scala BooleanNode API **程序包：com.amazonaws.services.glue.types** ## BooleanNode case 类 **BooleanNode** ``` case class BooleanNode extends ScalarNode(value, TypeCode.BOOLEAN) ( value : Boolean ) ``` ### BooleanNode val 字段 + `ordering` ### BooleanNode def 方法 ``` def equals( other : Any ) ``` # AWS Glue Scala ByteNode API **程序包：com.amazonaws.services.glue.types** ## ByteNode case 类 **ByteNode** ``` case class ByteNode extends ScalarNode(value, TypeCode.BYTE) ( value : Byte ) ``` ### ByteNode val 字段 + `ordering` ### ByteNode def 方法 ``` def equals( other : Any ) ``` # AWS Glue Scala DateNode API **程序包：com.amazonaws.services.glue.types** ## DateNode case 类 **DateNode** ``` case class DateNode extends ScalarNode(value, TypeCode.DATE) ( value : Date ) ``` ### DateNode val 字段 + `ordering` ### DateNode def 方法 ``` def equals( other : Any ) ``` ``` def this( value : Int ) ``` # AWS Glue Scala DecimalNode API **程序包：com.amazonaws.services.glue.types** ## DecimalNode case 类 **DecimalNode** ``` case class DecimalNode extends ScalarNode(value, TypeCode.DECIMAL) ( value : BigDecimal ) ``` ### DecimalNode val 字段 + `ordering` ### DecimalNode def 方法 ``` def equals( other : Any ) ``` ``` def this( value : Decimal ) ``` # AWS Glue Scala DoubleNode API **程序包：com.amazonaws.services.glue.types** ## DoubleNode case 类 **DoubleNode** ``` case class DoubleNode extends ScalarNode(value, TypeCode.DOUBLE) ( value : Double ) ``` ### DoubleNode val 字段 + `ordering` ### DoubleNode def 方法 ``` def equals( other : Any ) ``` # AWS Glue Scala DynamicNode API **Topics** + [DynamicNode 类](#glue-etl-scala-apis-glue-types-dynamicnode-class) + [DynamicNode 对象](#glue-etl-scala-apis-glue-types-dynamicnode-object) **程序包：com.amazonaws.services.glue.types** ## DynamicNode 类 **DynamicNode** ``` class DynamicNode extends Serializable with Cloneable ``` ### DynamicNode def 方法 ``` def getValue : Any ``` 获取无格式值并绑定到当前记录： ``` def nodeType : TypeCode ``` ``` def toJson : String ``` 调试方法： ``` def toRow( schema : Schema, options : Map[String, ResolveOption] ) : Row ``` ``` def typeName : String ``` ## DynamicNode 对象 **DynamicNode** ``` object DynamicNode ``` ### DynamicNode def 方法 ``` def quote( field : String, useQuotes : Boolean ) : String ``` ``` def quote( node : DynamicNode, useQuotes : Boolean ) : String ``` # EvaluateDataQuality 类 | | | --- | | AWS Glue Data Quality 目前为 AWS Glue 的预览版，可能会发生变化。 | **程序包：com.amazonaws.services.glue.dq** ``` object EvaluateDataQuality ``` ## Def apply ``` def apply(frame: DynamicFrame, ruleset: String, publishingOptions: JsonOptions = JsonOptions.empty): DynamicFrame ``` 根据 `DynamicFrame` 评估数据质量规则集，并返回一个包含评估结果的新 `DynamicFrame`。要了解有关 AWS Glue 数据质量的更多信息，请参阅[AWS Glue Data Quality](glue-data-quality.md)。 + `frame` 表示您想要评估的数据质量的 `DynamicFrame`。 + `ruleset` – 字符串格式的数据质量定义语言（DQDL）规则集。要了解有关 DQDL 的更多信息，请参阅[数据质量定义语言（DQDL）引用](dqdl.md)指南。 + `publishingOptions` – 一个字典，用于为发布评估结果和指标指定以下选项： + `dataQualityEvaluationContext` – 一个字符串，用于指定 AWS Glue 应在哪个命名空间下发布 Amazon CloudWatch 指标和数据质量结果。汇总指标显示在 CloudWatch 中，而完整结果显示在 AWS Glue Studio 界面中。 + 必需：否 + 默认值：`default_context` + `enableDataQualityCloudWatchMetrics` – 指定是否应将数据质量评估结果发布到 CloudWatch。您可以使用 `dataQualityEvaluationContext` 选项为指标指定命名空间。 + 必需：否 + 默认值：False + `enableDataQualityResultsPublishing` – 指定是否应在 AWS Glue Studio 界面的 **Data Quality**（数据质量）选项卡上显示数据质量结果。 + 必需：否 + 默认值：true + `resultsS3Prefix` – 指定 AWS Glue 可以写入数据质量评估结果的 Amazon S3 位置。 + 必需：否 + 默认值：""（空字符串） ## 示例以下示例代码演示了在执行 `SelectFields` 转换之前如何评估 `DynamicFrame` 数据质量。该脚本在尝试转换之前会验证所有数据质量规则是否均已通过。 ``` import com.amazonaws.services.glue.GlueContext import com.amazonaws.services.glue.MappingSpec import com.amazonaws.services.glue.errors.CallSite import com.amazonaws.services.glue.util.GlueArgParser import com.amazonaws.services.glue.util.Job import com.amazonaws.services.glue.util.JsonOptions import org.apache.spark.SparkContext import scala.collection.JavaConverters._ import com.amazonaws.services.glue.dq.EvaluateDataQuality object GlueApp { def main(sysArgs: Array[String]) { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) // @params: [JOB_NAME] val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME").toArray) Job.init(args("JOB_NAME"), glueContext, args.asJava) // Create DynamicFrame with data val Legislators_Area = glueContext.getCatalogSource(database="legislators", tableName="areas_json", transformationContext="S3bucket_node1").getDynamicFrame() // Define data quality ruleset val DQ_Ruleset = """ Rules = [ColumnExists "id"] """ // Evaluate data quality val DQ_Results = EvaluateDataQuality.apply(frame=Legislators_Area, ruleset=DQ_Ruleset, publishingOptions=JsonOptions("""{"dataQualityEvaluationContext": "Legislators_Area", "enableDataQualityMetrics": "true", "enableDataQualityResultsPublishing": "true"}""")) assert(DQ_Results.filter(_.getField("Outcome").contains("Failed")).count == 0, "Failing DQ rules for Legislators_Area caused the job to fail.") // Script generated for node Select Fields val SelectFields_Results = Legislators_Area.selectFields(paths=Seq("id", "name"), transformationContext="Legislators_Area") Job.commit() } } ``` # AWS Glue Scala FloatNode API **程序包：com.amazonaws.services.glue.types** ## FloatNode case 类 **FloatNode** ``` case class FloatNode extends ScalarNode(value, TypeCode.FLOAT) ( value : Float ) ``` ### FloatNode val 字段 + `ordering` ### FloatNode def 方法 ``` def equals( other : Any ) ``` # FillMissingValues 类 **程序包：com.amazonaws.services.glue.ml** ``` object FillMissingValues ``` ## Def apply ``` def apply(frame: DynamicFrame, missingValuesColumn: String, outputColumn: String = "", transformationContext: String = "", callSite: CallSite = CallSite("Not provided", ""), stageThreshold: Long = 0, totalThreshold: Long = 0): DynamicFrame ``` 填充指定列中的动态帧缺失值，并在新列中返回包含估计值的新帧。对于没有缺失值的行，指定列的值将复制到新列。 + `frame` – 要填充缺失值的 DynamicFrame。必需。 + `missingValuesColumn` – 包含缺少值的列（`null` 值和空字符串）。必需。 + `outputColumn` – 新列的名称，该列将包含值缺失的所有行的估计值。可选；默认为后缀为 `"_filled"` 的 `missingValuesColumn` 值。 + `transformationContext` – 用于标识状态信息的唯一字符串（可选）。 + `callSite` – 用于为错误报告提供上下文信息（可选）。 + `stageThreshold` – 在转换出错之前可能在其中发生的最大错误数（可选；默认值为零）。 + `totalThreshold` – 在处理出错之前可能全面发生的最大错误数（可选；默认值为零）。返回带附加列的新动态帧，该列包含带缺失值的行的估计值和其他行的当前值。 # FindMatches 类 **程序包：com.amazonaws.services.glue.ml** ``` object FindMatches ``` ## Def apply ``` def apply(frame: DynamicFrame, transformId: String, transformationContext: String = "", callSite: CallSite = CallSite("Not provided", ""), stageThreshold: Long = 0, totalThreshold: Long = 0, enforcedMatches: DynamicFrame = null): DynamicFrame, computeMatchConfidenceScores: Boolean ``` 在输入框架中查找匹配项并返回一个新框架，其中包含每个匹配组唯一 ID 的新列。 + `frame` – 要在其中查找匹配项的 DynamicFrame。必需。 + `transformId` – 与要应用于输入框架的 FindMatches 转换关联的唯一 ID。必需。 + `transformationContext` – 此 `DynamicFrame` 的标识符。`transformationContext` 用作跨运行保存的任务书签状态的密钥。可选。 + `callSite` – 用于为错误报告提供上下文信息。在从 Python 调用时，会自动设置这些值。可选。 + `stageThreshold` – 在引发异常之前允许的来自此 `DynamicFrame` 计算的最大错误记录数，不包括以前的 `DynamicFrame` 中存在的记录。可选。默认值为 0。 + `totalThreshold` – 引发异常之前的最大错误记录总数，包括以前的帧中的记录。可选。默认值为 0。 + `enforcedMatches` – 强制匹配的帧。可选。默认为 `null`。 + `computeMatchConfidenceScores` — 布尔值，指示是否为每组匹配记录计算置信度得分。可选。默认值为 false。返回一个新的动态框架，该框架具有分配给每组匹配记录的唯一标识符。 # FindIncrementalMatches 类 **程序包：com.amazonaws.services.glue.ml** ``` object FindIncrementalMatches ``` ## Def apply ``` apply(existingFrame: DynamicFrame, incrementalFrame: DynamicFrame, transformId: String, transformationContext: String = "", callSite: CallSite = CallSite("Not provided", ""), stageThreshold: Long = 0, totalThreshold: Long = 0, enforcedMatches: DynamicFrame = null): DynamicFrame, computeMatchConfidenceScores: Boolean ``` 查找现有帧和增量帧之间的匹配项，并返回一个新帧，其中包含每个匹配组唯一 ID 的新列。 + `existingframe` – 已为每个组分配了匹配 ID 的现有帧。必需。 + `incrementalframe` – 用于查找与现有帧匹配的增量帧。必需。 + `transformId` – 与要应用于输入帧的 FindIncrementalMatches 转换相关联的唯一 ID。必需。 + `transformationContext` – 此 `DynamicFrame` 的标识符。`transformationContext` 用作跨运行保存的任务书签状态的密钥。可选。 + `callSite` – 用于为错误报告提供上下文信息。在从 Python 调用时，会自动设置这些值。可选。 + `stageThreshold` – 在引发异常之前允许的来自此 `DynamicFrame` 计算的最大错误记录数，不包括以前的 `DynamicFrame` 中存在的记录。可选。默认值为 0。 + `totalThreshold` – 引发异常之前的最大错误记录总数，包括以前的帧中的记录。可选。默认值为 0。 + `enforcedMatches` – 强制匹配的帧。可选。默认为 `null`。 + `computeMatchConfidenceScores` — 布尔值，指示是否为每组匹配记录计算置信度得分。可选。默认值为 false。返回一个新的动态框架，该框架具有分配给每组匹配记录的唯一标识符。 # AWS Glue Scala IntegerNode API **程序包：com.amazonaws.services.glue.types** ## IntegerNode case 类 **IntegerNode** ``` case class IntegerNode extends ScalarNode(value, TypeCode.INT) ( value : Int ) ``` ### IntegerNode val 字段 + `ordering` ### IntegerNode def 方法 ``` def equals( other : Any ) ``` # AWS Glue Scala LongNode API **程序包：com.amazonaws.services.glue.types** ## LongNode case 类 **LongNode** ``` case class LongNode extends ScalarNode(value, TypeCode.LONG) ( value : Long ) ``` ### LongNode val 字段 + `ordering` ### LongNode def 方法 ``` def equals( other : Any ) ``` # AWS Glue Scala MapLikeNode API **程序包：com.amazonaws.services.glue.types** ## MapLikeNode 类 **MapLikeNode** ``` class MapLikeNode extends DynamicNode ( value : mutable.Map[String, DynamicNode] ) ``` ### MapLikeNode def 方法 ``` def clear : Unit ``` ``` def get( name : String ) : Option[DynamicNode] ``` ``` def getValue ``` ``` def has( name : String ) : Boolean ``` ``` def isEmpty : Boolean ``` ``` def put( name : String, node : DynamicNode ) : Option[DynamicNode] ``` ``` def remove( name : String ) : Option[DynamicNode] ``` ``` def toIterator : Iterator[(String, DynamicNode)] ``` ``` def toJson : String ``` ``` def toJson( useQuotes : Boolean ) : String ``` **示例：**鉴于此 JSON： ``` {"foo": "bar"} ``` 如果 `useQuotes == true`，则 `toJson` 会生成 `{"foo": "bar"}`。如果 `useQuotes == false`，则 `toJson` 会生成 `{foo: bar}` @return。 # AWS Glue Scala MapNode API **程序包：com.amazonaws.services.glue.types** ## MapNode case 类 **MapNode** ``` case class MapNode extends MapLikeNode(value) ( value : mutable.Map[String, DynamicNode] ) ``` ### MapNode def 方法 ``` def clone ``` ``` def equals( other : Any ) ``` ``` def hashCode : Int ``` ``` def nodeType ``` ``` def this ``` # AWS Glue Scala NullNode API **Topics** + [NullNode 类](#glue-etl-scala-apis-glue-types-nullnode-class) + [NullNode case 对象](#glue-etl-scala-apis-glue-types-nullnode-case-object) **程序包：com.amazonaws.services.glue.types** ## NullNode 类 **NullNode** ``` class NullNode ``` ## NullNode case 对象 **NullNode** ``` case object NullNode extends NullNode ``` # AWS Glue Scala ObjectNode API **Topics** + [ObjectNode 对象](#glue-etl-scala-apis-glue-types-objectnode-object) + [ObjectNode case 类](#glue-etl-scala-apis-glue-types-objectnode-case-class) **程序包：com.amazonaws.services.glue.types** ## ObjectNode 对象 **ObjectNode** ``` object ObjectNode ``` ### ObjectNode def 方法 ``` def apply( frameKeys : Set[String], v1 : mutable.Map[String, DynamicNode], v2 : mutable.Map[String, DynamicNode], resolveWith : String ) : ObjectNode ``` ## ObjectNode case 类 **ObjectNode** ``` case class ObjectNode extends MapLikeNode(value) ( val value : mutable.Map[String, DynamicNode] ) ``` ### ObjectNode def 方法 ``` def clone ``` ``` def equals( other : Any ) ``` ``` def hashCode : Int ``` ``` def nodeType ``` ``` def this ``` # AWS Glue Scala ScalarNode API **Topics** + [ScalarNode 类](#glue-etl-scala-apis-glue-types-scalarnode-class) + [ScalarNode 对象](#glue-etl-scala-apis-glue-types-scalarnode-object) **程序包：com.amazonaws.services.glue.types** ## ScalarNode 类 **ScalarNode** ``` class ScalarNode extends DynamicNode ( value : Any, scalarType : TypeCode ) ``` ### ScalarNode def 方法 ``` def compare( other : Any, operator : String ) : Boolean ``` ``` def getValue ``` ``` def hashCode : Int ``` ``` def nodeType ``` ``` def toJson ``` ## ScalarNode 对象 **ScalarNode** ``` object ScalarNode ``` ### ScalarNode def 方法 ``` def apply( v : Any ) : DynamicNode ``` ``` def compare( tv : Ordered[T], other : T, operator : String ) : Boolean ``` ``` def compareAny( v : Any, y : Any, o : String ) ``` ``` def withEscapedSpecialCharacters( jsonToEscape : String ) : String ``` # AWS Glue Scala ShortNode API **程序包：com.amazonaws.services.glue.types** ## ShortNode case 类 **ShortNode** ``` case class ShortNode extends ScalarNode(value, TypeCode.SHORT) ( value : Short ) ``` ### ShortNode val 字段 + `ordering` ### ShortNode def 方法 ``` def equals( other : Any ) ``` # AWS Glue Scala StringNode API **程序包：com.amazonaws.services.glue.types** ## StringNode case 类 **StringNode** ``` case class StringNode extends ScalarNode(value, TypeCode.STRING) ( value : String ) ``` ### StringNode val 字段 + `ordering` ### StringNode def 方法 ``` def equals( other : Any ) ``` ``` def this( value : UTF8String ) ``` # AWS Glue Scala TimestampNode API **程序包：com.amazonaws.services.glue.types** ## TimestampNode case 类 **TimestampNode** ``` case class TimestampNode extends ScalarNode(value, TypeCode.TIMESTAMP) ( value : Timestamp ) ``` ### TimestampNode val 字段 + `ordering` ### TimestampNode def 方法 ``` def equals( other : Any ) ``` ``` def this( value : Long ) ``` # AWS Glue Scala GlueArgParser API **程序包：com.amazonaws.services.glue.util** ## GlueArgParser 对象 **GlueArgParser** ``` object GlueArgParser ``` 这与 `AWSGlueDataplanePython` 程序包中的 `utils.getResolvedOptions` 的 Python 版本严格一致。 ### GlueArgParser def 方法 ``` def getResolvedOptions( args : Array[String], options : Array[String] ) : Map[String, String] ``` ``` def initParser( userOptionsSet : mutable.Set[String] ) : ArgumentParser ``` **Example 检索传递到作业的参数** 要检索作业参数，您可以使用 `getResolvedOptions` 方法。考虑以下示例，它检索名为 `aws_region` 的作业参数。 ``` val args = GlueArgParser.getResolvedOptions(sysArgs, Seq("JOB_NAME","aws_region").toArray) Job.init(args("JOB_NAME"), glueContext, args.asJava) val region = args("aws_region") println(region) ``` # AWS Glue Scala 作业 API **程序包：com.amazonaws.services.glue.util** ## Job 对象 **任务** ``` object Job ``` ### Job def 方法 ``` def commit ``` ``` def init( jobName : String, glueContext : GlueContext, args : java.util.Map[String, String] = Map[String, String]().asJava ) : this.type ``` ``` def init( jobName : String, glueContext : GlueContext, endpoint : String, args : java.util.Map[String, String] ) : this.type ``` ``` def isInitialized ``` ``` def reset ``` ``` def runId ```