# AWS Glue Scala ライブラリの API AWS Glue では、抽出、変換、ロード (ETL) ジョブのスクリプト化に使用する PySpark Scala 言語の拡張機能がサポートされています。以下のセクションでは、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) **パッケージ: 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 ``` ## ケースクラス 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) **パッケージ: 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 トレイト **パッケージ: 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 DataFrames との相互変換や、DataFrames が提供する多くの分析オペレーションをサポートしています。以下のパラメータは、`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` – ソース列で大文字と小文字を区別して扱うかどうかを指定。このパラメータを false に設定すると、AWS Glue Data Catalog のように大文字と小文字を区別しないストアと統合する場合に役立ちます。マッピングのシーケンスに基づく選択列、プロジェクト列、およびキャスト列。各マッピングは、ソース列/タイプとターゲット列/タイプで構成されます。マッピングは、4 タプル (`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 ) ``` すべての NULL 列が削除された新しい `DynamicFrame` を返します。 **注記** `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 ``` "`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` との等結合を行った結果を返します。 ## 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` – Relationalize のオプションと設定。現在使用されていません。すべてのネストされた構造をフラット化し、配列を個別のテーブルにピボットします。このオペレーションでは、リレーショナルデータベースに取り込むための深くネストされたデータを準備できます。ネストされた構造体は、[ネスト解除](#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) }}} ``` これにより、2 つのテーブルが生成されます。最初のテーブルは "people" という名前で、内容は以下のとおりです。 ``` {{{ {"name": "Nancy", "age": 47, "friends": 1} {"name": "Stephanie", "age": 28, "friends": 2} {"name": "Nathan", "age": 54, "friends": 3) }}} ``` ここで、友人の配列は自動生成された結合キーに置き換えられています。別のテーブルは `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 テーブル。 1 つ以上の `ChoiceType` をより限定されたタイプに置き換えて新しい `DynamicFrame` を返します。 `resolveChoice` を使用するには 2 つの方法があります。最初の方法では、特定の列のシーケンスと解決方法を指定します。これらは (列、アクション) ペアで構成されたタプルとして指定します。以下のアクションを指定できます。 + `cast:type` – すべての値を指定した型にキャストしようとします。 + `make_cols` – それぞれの異なるタイプを `columnName_type` という名前の列に変換します。 + `make_struct` – 列を各区別型のキーを持つ構造体に変換します。 + `project:type` - 指定した型の値のみを保持します。 `resolveChoice` のもう 1 つのモードでは、すべての `ChoiceType` に対して単一の解決策を指定します。このモードは、実行前に `ChoiceType` の完全なリストが不明な場合に使用できます。このモードでは、上記のアクションに加えて、以下のアクションもサポートされています。 + `match_catalog``ChoiceType` – 指定したカタログテーブルの対応するタイプへの各のキャストを試行します。 **例:** `user.id` 列を解決するために int にキャストし、`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` のレコードにあるすべてのフィールドが含まれます。しかし、その他のフィールドが含まれる場合がまれにあります。[ネスト解除](#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` - 出力する行の数。この `DynamicFrame` の行を JSON 形式で出力します。 ## Def simplifyDDBJson DynamoDB は AWS Glue DynamoDB エクスポートコネクタを使用してエクスポートし、特殊なネスト構造の JSON ファイルが生成されます。詳細については、「[データオブジェクト](https://docs.aws.amazon.com/amazondynamodb/latest/developerguide/S3DataExport.Output.html)」を参照してください。`simplifyDDBJson`このタイプのデータの DynamicFrame にあるネスト化された列を単純化し、新しい単純化された DynamicFrame を返します。リストタイプに複数タイプまたは Map タイプが含まれている場合、リストの要素は単純化されません。このメソッドは、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` を返します。デフォルトでは、`path` で指定した場所に任意の 100 レコードを書き込みます。この動作は、`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` に含めるパス。 2 つの `DynamicFrame` のシーケンスを返します。`DynamicFrame` の 1 つ目には指定したパスが含まれ、2 つ目には他のすべての列が含まれます。 **例** この例では、`persons` Glue Data Catalog の `legislators` データベースの AWS テーブルから作成した DynamicFrame を 2 つに分割し、指定したフィールドを 1 つ目の DynamicFrame に、残りのフィールドを 2 つ目の 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` - 比較に使用する演算子。 2 つの `DynamicFrame` のシーケンスを返します。1 つ目には述語が true の行を含め、2 つ目には述語が false の行を含めます。述語を指定するには 3 つのシーケンスを使用します。"`paths`" には、ネストされている可能性がある列の名前を含め、"'`values`' には、比較する定数値を含め、"`operators`" には、比較に使用する演算子を含めます。3 つすべてのシーケンスを同じ長さにする必要があります。`n` 演算子では、`n` 番目の列を `n` 番目の値と比較します。各演算子は "`!=`"、"`=`"、"`<=`"、"`<`"、"`>=`"、"`>`" のいずれかにする必要があります。以下の呼び出しの例では、`DynamicFrame` を分割し、1 つ目の出力フレームには米国の 65 才を超える人々のレコード、2 つ目には他のすべてのレコードを含めています。 ``` {{{ 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 }}} ``` 次の呼び出しでアドレス構造体がネスト解除されます。 ``` {{{ 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 unnest を呼び出し、パーティションの数を表示します。 ``` 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] ``` `DynamicNode` のオプションとして指定した `path` でフィールドの値を取得します。フィールドが存在する場合は `scala.Some Some`([DynamicNode](glue-etl-scala-apis-glue-types-dynamicnode.md)) を返し、それ以外の場合は `scala.None.None` を返します。 ## def getField ``` def getField( path : String ) : Option[Any] ``` `DynamicNode` のオプションとして指定した `path` でフィールドの値を取得します。 `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` は、Amazon Simple Storage Service (Amazon S3) や、AWS Glue Data Catalog、JDBC その他との間で、[DynamicFrame](glue-etl-scala-apis-glue-dynamicframe.md) の読み取りと書き込みを行うためのエントリポイントです。このクラスが提供するユーティリティ関数によって [DataSource 特性](glue-etl-scala-apis-glue-datasource-trait.md) オブジェクトと [DataSink](glue-etl-scala-apis-glue-datasink-class.md) オブジェクトが作成され、これらのオブジェクトを使用して `DynamicFrame` を読み書きできます。また、ソースから作成されたパーティションの数がパーティションの最小しきい値 (デフォルトは 10) 未満の場合は、`DynamicFrame` で `GlueContext` を使用してパーティションのターゲット数 (デフォルトは 20) を設定することもできます。 ## def addIngestionTimeColumns ``` def addIngestionTimeColumns( df : DataFrame, timeGranularity : String = "") : dataFrame ``` 入力 `DataFrame` への取り込み時間列 (`ingest_year`、`ingest_month`、`ingest_day`、`ingest_hour`、`ingest_minute` など) を追加します。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` が必要です。 + 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 ``` Data Catalog で定義されているテーブル内の指定した場所に書き込みを行う [DataSink](glue-etl-scala-apis-glue-datasink-class.md) を作成します。 + `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 ``` Data Catalog のテーブル定義からデータを読み取る [DataSource 特性](glue-etl-scala-apis-glue-datasource-trait.md) を作成します。 + `database` – Data Catalog 内のデータベース名。 + `tableName` – Data Catalog 内のテーブル名。 + `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` – 現在アクセスされている 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 ``` Data Catalog 内の `Connection` オブジェクトに指定されている、JDBC データベースに対し書き込みを行う [DataSink](glue-etl-scala-apis-glue-datasink-class.md) を作成します。`Connection` オブジェクトには、URL、ユーザー名、パスワード、VPC、サブネット、セキュリティグループなど、JDBC シンクに接続するための情報があります。 + `catalogConnection` – 書き込み先の JDBC URL を含む Data Catalog の接続名。 + `options` – JDBC データストアへの書き込みに必要な追加情報を提供する、名前/値ペアを示す JSON 形式の文字列。これには、以下が含まれます。 + *dbtable* (必須) - JDBC テーブルの名前。データベース内でスキーマをサポートする JDBC データストアの場合、`schema.table-name` を指定します。スキーマを指定しない場合、デフォルトの「パブリック」スキーマが使用されます。次の例では、`test_db` データベースの `test` という名前のスキーマおよび `test_table` という名前のテーブルを指すオプションパラメータを示します。 ``` options = JsonOptions("""{"dbtable": "test.test_table", "database": "test_db"}""") ``` + *database* (必須) - JDBC データベースの名前。 + その他のオプションはすべて SparkSQL JDBC ライターに直接渡されます。詳細については、「[Redshift data source for Spark](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 ``` Amazon Simple Storage Service (Amazon S3)、JDBC、AWS Glue Data Catalog、Apache Kafka、Amazon Kinesis Data Streams などの書き込み先にデータを書き込む [DataSink](glue-etl-scala-apis-glue-datasink-class.md) を作成します。 + `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 ``` Amazon S3、JDBC、Data Catalog、Apache Kafka、Amazon Kinesis Data Streams などの書き込み先にデータを書き込む [DataSink](glue-etl-scala-apis-glue-datasink-class.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` - 書き込み先に書き込むデータの形式。 + `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 Data Catalog などのソースからデータを読み取る [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 オブジェクトでは、DynamicFrames から作成した `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 commitTrac ``` def commitTransaction(transactionId: String, waitForCommit: Boolean): Boolean ``` 指定されたトランザクションをコミットしようとします。`commitTransaction` トランザクションのコミットが完了する前に戻ることがあります。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-CommitTransaction) API を内部的に呼び出します。 + `transactionId` — (文字列) コミットするトランザクション。 + `waitForCommit` — (ブール値) `commitTransaction` がすぐに戻るかどうか指定します。デフォルト値は True です。false の場合、`commitTransaction` はトランザクションがコミットされるまでポーリングし待機します。最大で 6 回の再試行でエクスポネンシャルバックオフを使用すると、待機時間は 1 分に制限されます。コミットが完了したかどうかを示すブール値を返します。 ## def cancelTransaction ``` def cancelTransaction(transactionId: String): Unit ``` 指定されたトランザクションをキャンセルしようとします。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-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 ) ``` 渡された `SparkContext` で `GlueContext` オブジェクトを作成します。最小限のパーティション数を 10 に設定し、ターゲットパーティション数を 20 に設定します。 + `sc` - `SparkContext`。戻り値は `GlueContext`。 ## def this ``` def this( sparkContext : JavaSparkContext ) ``` 渡された `JavaSparkContext` で `GlueContext` オブジェクトを作成します。最小限のパーティション数を 10 に設定し、ターゲットパーティション数を 20 に設定します。 + `sparkContext` - `JavaSparkContext`。戻り値は `GlueContext`。 # MappingSpec **パッケージ: 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` からの拡張であるため、`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 ケースクラス](#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 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 ケースクラス **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 ケースクラス **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 ケースクラス **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 ケースクラス **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 ケースクラス **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 ケースクラス **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 ケースクラス **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 Data Quality の詳細については、「[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 ケースクラス **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` — エラーで終了する前に変換で許容される、エラーの最大発生数 (オプション、デフォルト値は 0)。 + `totalThreshold` — エラーによる終了が処理される前に、全体で許容されるエラーの最大発生数 (オプション、デフォルト値は 0)。値が欠落した行のための推定値を含む (追加された) 1 つの列、さらに他の行の現在値を含む、新しい動的フレームを返します。 # 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 ケースクラス **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 ケースクラス **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 ケースクラス **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) **パッケージ: 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) **パッケージ: 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) **パッケージ: 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 ケースクラス **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 ケースクラス **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 ケースクラス **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 Job 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 ```