本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。 # 在 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) + [套用](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) + [DataSource 特徵](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) + [Join](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) + [Show](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) + [複製](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) + [套用](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 案例類別](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) + [套用](glue-etl-scala-apis-glue-mappingspec.md#glue-etl-scala-apis-glue-mappingspec-object-defs-apply-1) + [套用](glue-etl-scala-apis-glue-mappingspec.md#glue-etl-scala-apis-glue-mappingspec-object-defs-apply-2) + [套用](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 案例類別](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 案例類別](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 案例類別](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 案例類別](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 案例類別](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 案例類別](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 案例類別](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 案例類別](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 案例類別](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) + [套用](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) + [套用](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) + [套用](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 案例類別](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 案例類別](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 案例類別](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 案例物件](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 案例類別](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 案例類別](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 案例類別](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 案例類別](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) + [任務物件](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) 程式並視需要修改，再將其指派到任務。或者，您也可以從頭開始撰寫自己的程式。如需更多資訊，請參閱 [在中設定 Spark 任務的任務屬性 AWS Glue](add-job.md)。AWS Glue 會在執行相關的任務前在伺服器編譯 Scala 程式。為了確保您的程式編譯無誤且如預期般執行，您必須在任務中執行該程式前，在 REPL (Read-Eval-Print Loop) 或 Jupyter 筆記本的開發端點上將其載入並測試。由於編譯處理會在伺服器上進行，您將無法詳細查看在其中發生的任何問題。 ## 在開發端點上使用 Jupyter 筆記本測試 Scala ETL 程式若要在 AWS Glue 開發端點上測試 Scala 程式，請如 [新增開發端點](add-dev-endpoint.md) 所述來設定開發端點。接著，將其連線至在本機電腦或 Amazon EC2 筆記本伺服器遠端執行中的 Jupyter 筆記本。若要安裝本機版本的 Jupyter 筆記本，請遵循 [教學課程：JupyterLab 中的 Jupyter 筆記本](dev-endpoint-tutorial-local-jupyter.md) 中的說明進行。在筆記本上執行 Scala 程式碼與執行 PySpark 程式碼之間的唯一差別，是您應使用以下項目以在筆記本上開始每一段落： ``` %spark ``` 這可防止筆記本伺服器將 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，使用來自 Data Catalog 的結構描述剖析資料串流，將串流聯結至 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) + [DataSource 特徵](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) + [案例類別 ChoiceOptionWithResolver](#glue-etl-scala-apis-glue-choiceoptionwithresolver-case-class) + [案例類別 MatchCatalogSchemaChoiceOption](#glue-etl-scala-apis-glue-matchcatalogschemachoiceoption-case-class) **Package: com.amazonaws.services.glue** ## ChoiceOption 特徵 ``` trait ChoiceOption extends Serializable ``` ## ChoiceOption 物件 **ChoiceOption** ``` object ChoiceOption ``` 用來解析所有 `ChoiceType` 節點 (`DynamicFrame` 中) 適用選擇的一般策略。 + `val CAST` + `val MAKE_COLS` + `val MAKE_STRUCT` + `val MATCH_CATALOG` + `val PROJECT` ### Def apply ``` def apply(choice: String): ChoiceOption ``` ## 案例類別 ChoiceOptionWithResolver ``` case class ChoiceOptionWithResolver(name: String, choiceResolver: ChoiceResolver) extends ChoiceOption {} ``` ## 案例類別 MatchCatalogSchemaChoiceOption ``` case class MatchCatalogSchemaChoiceOption() extends ChoiceOption {} ``` # Abstract 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) **Package: 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 DataSource 特徵 **Package: 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 **Package: 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 類別 **Package: 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 DataFrames 的相互轉換，以整合現有程式碼以及 DataFrames 提供許多分析操作。以下參數在許多 AWS Glue 轉換之間共用以建構 `DynamicFrame`： + `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` – 是否將來源欄位視為區分大小寫。將此設定為 false 可能有助於與不區分大小寫的存放區整合，例如 AWS Glue Data Catalog。依據映射序列選取、投影及投射欄位。每個映射皆由來源欄位和類型以及目標欄位和類型所組成。映射可能會指定為四元組 (`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` 並移除所有 null 欄位。 **注意** 這只會移除 `NullType` 類型的欄位。其他欄位中的個別 null 值不會被移除或修改。 ## 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 ``` 建構新的 `DynamicFrame`，其中僅包含函數「`f`」傳回 `true` 的那些記錄。篩選條件函數「`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` 執行對等聯結的結果。 ## Def map ``` def map( f : DynamicRecord => DynamicRecord, errorMsg : String = "", transformationContext : String = "", callSite : CallSite = CallSite("Not provided", ""), stageThreshold : Long = 0, totalThreshold : Long = 0 ) : DynamicFrame ``` 傳回藉由將指定函數「`f`」套用至此 `DynamicFrame` 中各個記錄而建構的新 `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` – A `Long`。在給定轉換中的錯誤數量，其處理需要輸出錯誤。 + `totalThreshold` – A `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`。您可以使用這個方法來重新命名巢狀欄位。例如，以下程式碼會將地址結構中的 `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` – 套用到所有未在規格序列中列出之 `ChoiceType` 欄位的動作。 + `database` – 搭配 `match_catalog` 動作使用的 Data Catalog 資料庫。 + `tableName` – 搭配 `match_catalog` 動作使用的 Data 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 DynamoDB 會使用 AWS Glue 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` – 以 `s3://bucket//path` 格式將輸出寫入至 Amazon S3 中的路徑。 + `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 Data Catalog 中`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 分隔符號。根據指定的格式，剖析嵌入字串或二進位欄位。剖析的欄位是具有原始資料欄名稱結構的巢狀欄位。例如，假設您有 CSV 檔案與內嵌 JSON 欄位。 ``` 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 }}} ``` 以下呼叫將會解巢狀地址結構。 ``` {{{ 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 ``` 解除專屬於 DynamoDB JSON 結構中 `DynamicFrame` 內的巢狀欄的巢狀化，並傳回新的解巢狀 `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 un巢狀，以及列印分割區數量： ``` 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 物件 **Package: 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) **Package: 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` 的選項。如果欄位存在，傳回 `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` 的選項。傳回 `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 **Package: 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 Data Catalog 、JDBC 等的進入點。此類別提供公用程式函數來建立 [DataSource 特徵](glue-etl-scala-apis-glue-datasource-trait.md) 和 [DataSink](glue-etl-scala-apis-glue-datasink-class.md) 物件，從而用於讀取和寫入 `DynamicFrame`。如果從來源建立的分割區數低於分割區的閾值下限 (預設 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 為目標的 Data Catalog 資料表時，此函數會在 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`。此函數僅適用於 Glue AWS 串流來源。 + `connectionType` – 串流連線類型。有效值包括 `kinesis` 與 `kafka`。 + `connectionOptions` – 連線選項，這些選項對於 Kinesis 和 Kafka 而言是不同的。您可以在 [AWS Glue for Spark 中 ETL 的連線類型和選項](aws-glue-programming-etl-connect.md) 中找到每個串流資料來源的所有連線選項清單。請注意串流連線選項的下列不同處： + Kinesis 串流來源需要 `streamARN`、`startingPosition`、`inferSchema` 以及 `classification`。 + Kafka 串流來源需要 `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)，以便寫入 Data Catalog 中定義之資料表中指定的位置。 + `database` — Data Catalog 中的資料庫名稱。 + `tableName` — Data Catalog 中的資料表名稱。 + `redshiftTmpDir` — 要與特定資料目的地搭配使用的臨時暫存目錄。設定為預設為空值。 + `transformationContext` — 與由任務書籤使用之目的地關聯的轉換內容。設定為預設為空值。 + `additionalOptions` – 提供給 AWS Glue 的額外選項。 + `catalogId` — 要存取之 Data Catalog 的目錄 ID (帳戶 ID)。為 null 時，會使用發起人的預設帳戶 ID。傳回 `DataSink`。 ## def getCatalogSource ``` def getCatalogSource( database : String, tableName : String, redshiftTmpDir : String = "", transformationContext : String = "" pushDownPredicate : String = " " additionalOptions: JsonOptions = JsonOptions.empty, catalogId: String = null ) : DataSource ``` 建立 [DataSource 特徵](glue-etl-scala-apis-glue-datasource-trait.md)，以便從 Data Catalog 中的資料表定義中讀取資料。 + `database` — Data Catalog 中的資料庫名稱。 + `tableName` — Data Catalog 中的資料表名稱。 + `redshiftTmpDir` — 要與特定資料目的地搭配使用的臨時暫存目錄。設定為預設為空值。 + `transformationContext` — 與由任務書籤使用之目的地關聯的轉換內容。設定為預設為空值。 + `pushDownPredicate` – 篩選分割區，而無需列出和讀取資料集中的所有檔案。如需詳細資訊，請參閱[使用 pushdown 述詞預先篩選](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` — 要存取之 Data Catalog 的目錄 ID (帳戶 ID)。為 null 時，會使用發起人的預設帳戶 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 ``` 建立 [DataSink](glue-etl-scala-apis-glue-datasink-class.md)，以便寫入 Data Catalog 中 `Connection` 物件所指定的 JDBC 資料庫。此 `Connection` 物件擁有用來對 JDBC 目的地連線的資訊 (包括 URL、使用者名稱、密碼、VPC、子網路和安全群組)。 + `catalogConnection` — Data Catalog 中的連線名稱，其中包含要做為寫入目的地之 JDBC URL。 + `options` — JSON 名稱值組的字串，可提供寫入 JDBC 資料存放區所需的其他資訊。其中包含： + *dbtable* (必要) — JDBC 資料表的名稱。若是支援資料庫內結構描述的 JDBC 資料存放區，請指定 `schema.table-name`。如果未提供結構描述，則會使用預設的 "public" 結構描述。以下範例說明 options 參數，它會指向資料庫 `test_db` 中名為 `test` 的結構描述和名為 `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` — 要存取之 Data Catalog 的目錄 ID (帳戶 ID)。為 null 時，會使用發起人的預設帳戶 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 Data Catalog 等目的地，或 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、Data Catalog、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 或 Glue Data Catalog AWS 等來源[DataSource 特徵](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`。 Kafka 串流來源需要以下連線選項：`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 ``` 建立從 Amazon S3、JDBC 或 AWS Glue Data Catalog 等來源[DataSource 特徵](glue-etl-scala-apis-glue-datasource-trait.md)讀取資料的，也會設定存放在來源中的資料格式。 + `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 物件以將資料表與 UDF 註冊為與從 DynamicFrames 建立的 `DataFrame` 搭配使用。傳回 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。如為 False，`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 此 ``` def this( sc : SparkContext, minPartitions : Int, targetPartitions : Int ) ``` 使用指定的 `SparkContext`、最小分割區和分割區目標來建立 `GlueContext` 物件。 + `sc` — `SparkContext`。 + `minPartitions` — 分割區最小數。 + `targetPartitions` — 分割區目標數。傳回 `GlueContext`。 ## def 此 ``` def this( sc : SparkContext ) ``` 透過提供的 `SparkContext` 建立 `GlueContext` 物件。將分割區的最小值設為 10，目標分割區設為 20。 + `sc` — `SparkContext`。傳回 `GlueContext`。 ## def 此 ``` def this( sparkContext : JavaSparkContext ) ``` 透過提供的 `JavaSparkContext` 建立 `GlueContext` 物件。將分割區的最小值設為 10，目標分割區設為 20。 + `sparkContext` — `JavaSparkContext`。傳回 `GlueContext`。 # MappingSpec **Package: com.amazonaws.services.glue** ## MappingSpec 案例類別 ``` 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` 擴展，讓您可以處理任何 `Product4` (在 `applyMapping` 介面中)。 ## 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 案例類別](#glue-etl-scala-apis-glue-resolvespec-case-class) **Package: 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 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 **Package: com.amazonaws.services.glue.types** ## ArrayNode 案例類別 **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 **Package: com.amazonaws.services.glue.types** ## BinaryNode 案例類別 **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 **Package: com.amazonaws.services.glue.types** ## BooleanNode 案例類別 **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 **Package: com.amazonaws.services.glue.types** ## ByteNode 案例類別 **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 **Package: com.amazonaws.services.glue.types** ## DateNode 案例類別 **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 **Package: com.amazonaws.services.glue.types** ## DecimalNode 案例類別 **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 **Package: com.amazonaws.services.glue.types** ## DoubleNode 案例類別 **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) **Package: 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 Data Quality，請參閱 [AWS Glue 資料品質](glue-data-quality.md)。 + `frame` – 您要評估資料品質的 `DynamicFrame`。 + `ruleset` – 字串格式的資料品質定義語言 (DQDL) 規則集。若要進一步了解 DQDL，請參閱 [資料品質定義語言 (DQDL) 參考](dqdl.md) 指南。 + `publishingOptions` – 指定以下用於發佈評估結果和指標的選項的字典： + `dataQualityEvaluationContext` – 指定 Glue AWS 應發佈 Amazon CloudWatch 指標和資料品質結果的命名空間的字串。彙總指標會出現在 CloudWatch 中，而完整結果會出現在 AWS Glue Studio 界面中。 + 必要：否 + 預設值：`default_context` + `enableDataQualityCloudWatchMetrics` – 指定是否應將資料品質評估的結果發佈至 CloudWatch。您可以使用 `dataQualityEvaluationContext` 選項指定指標的命名空間。 + 必要：否 + 預設值：False + `enableDataQualityResultsPublishing` – 指定資料品質結果是否應顯示在 AWS Glue Studio 介面的 **Data Quality** (資料品質) 索引標籤上。 + 必要：否 + 預設值：True + `resultsS3Prefix` – 指定 Glue AWS 可以寫入資料品質評估結果的 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 **Package: com.amazonaws.services.glue.types** ## FloatNode 案例類別 **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` — 新欄的名稱，該欄將包含所有缺少值的列的估計值。選擇性；預設為 `missingValuesColumn` 的值，字尾為 `"_filled"`。 + `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` 中存在的記錄。選用。預設為零。 + `totalThreshold` — 在擲回例外狀況之前，最大的錯誤記錄總計 (包括之前框架的數量)。選用。預設為零。 + `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` 中存在的記錄。選用。預設為零。 + `totalThreshold` — 在擲回例外狀況之前，最大的錯誤記錄總計 (包括之前框架的數量)。選用。預設為零。 + `enforcedMatches` — 強制相符的框架。選用。預設值為 `null`。 + `computeMatchConfidenceScores` – 布林值，指出是否運算每個相符記錄群組的可信度分數。選用。預設值為 false。傳回具有指派給每個相符記錄群組之唯一識別碼的新動態框架。 # AWS Glue Scala IntegerNode API **Package: com.amazonaws.services.glue.types** ## IntegerNode 案例類別 **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 **Package: com.amazonaws.services.glue.types** ## LongNode 案例類別 **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 **Package: 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 **Package: com.amazonaws.services.glue.types** ## MapNode 案例類別 **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 案例物件](#glue-etl-scala-apis-glue-types-nullnode-case-object) **Package: com.amazonaws.services.glue.types** ## NullNode 類別 **NullNode** ``` class NullNode ``` ## NullNode 案例物件 **NullNode** ``` case object NullNode extends NullNode ``` # AWS Glue Scala ObjectNode API **Topics** + [ObjectNode 物件](#glue-etl-scala-apis-glue-types-objectnode-object) + [ObjectNode 案例類別](#glue-etl-scala-apis-glue-types-objectnode-case-class) **Package: 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 案例類別 **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) **Package: 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 **Package: com.amazonaws.services.glue.types** ## ShortNode 案例類別 **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 **Package: com.amazonaws.services.glue.types** ## StringNode 案例類別 **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 **Package: com.amazonaws.services.glue.types** ## TimestampNode 案例類別 **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 **Package: 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 **Package: com.amazonaws.services.glue.util** ## 任務物件 **任務** ``` 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 ```