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