# AWS Glue for Spark での入出力のデータ形式に関するオプション これらのページでは、AWS Glue for Spark でサポートされているデータ形式の機能のサポートと設定パラメータについて説明します。この情報の使用と適用可能性の説明については、以下を参照してください。 ## AWS Glue でのデータ形式間の機能のサポート 各データ形式が、異なる AWS Glue の機能をサポートする場合があります。次の一般的な機能は、形式のタイプによってサポートされる場合とサポートされない場合があります。データ形式についてのドキュメントを参照し、当社の機能を活用して要件を満たす方法を理解してください。 | | | | --- |--- | | 読み取り | AWS Glue では、コネクタなどのリソースを追加することなく、データ形式を認識し解釈できます。 | | 書き込み | AWS Glue では、リソースを追加することなく、この形式でデータを書き込むことができます。他の Spark 環境と同様に、サードパーティライブラリをジョブに含め、標準の Apache Spark 関数を使用してデータを書き込むことができます。ライブラリを含める方法の詳細については、「[AWS Glue での Python ライブラリの使用](aws-glue-programming-python-libraries.md)」を参照してください。 | | ストリーミングの読み取り | AWS Glue は、Apache Kafka、Amazon Managed Streaming for Apache Kafka、または Amazon Kinesis メッセージストリームから、このデータ形式を認識して解釈できます。ストリームは一貫した形式でデータを表示することを想定しているため、DataFrames として読み込まれます。 | | 小さなファイルのグループ化 | AWS Glue は、AWS Glue 変換を実行する際にファイルをグループ化し、各ノードに送信されるバッチを処理できます。これにより、大量の小さなファイルを扱うワークロードのパフォーマンスを大幅に向上できます。詳細については、「[大きなグループの入力ファイルの読み取り](grouping-input-files.md)」を参照してください。 | | ジョブのブックマーク | AWS Glue は、ジョブブックマークを使用して、ジョブの実行間で同じデータセットに対して同じ作業を実行する変換の進行状況を追跡できます。これにより、最後のジョブ実行以降の新しいデータに対してのみ作業を行う必要があるデータセットを含むワークロードのパフォーマンスを向上させることができます。詳細については、「[ジョブのブックマークを使用した処理済みデータの追跡](monitor-continuations.md)」を参照してください。 | ## AWS Glue でデータ形式を操作するために使用されるパラメータ 特定の AWS Glue 接続タイプは複数の `format` タイプをサポートしており、`GlueContext.write_dynamic_frame.from_options` のようなメソッドを使用する場合は、`format_options` オブジェクトを使用してデータ形式に関する情報を指定する必要があります。 + `s3` - 詳細については、AWS Glue: [S3 接続パラメータ](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) の「Connection types and options for ETL」(ETL の接続タイプとオプション) を参照してください。また、Python でのこの接続タイプを容易にするメソッド [create\$1dynamic\$1frame\$1from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-create_dynamic_frame_from_options) および [write\$1dynamic\$1frame\$1from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-write_dynamic_frame_from_options) と、対応する Scala メソッド [def getSourceWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSourceWithFormat) および [def getSinkWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSinkWithFormat) についてのドキュメントもご覧ください。 + `kinesis` - 詳細については、AWS Glue: [Kinesis 接続パラメータ](aws-glue-programming-etl-connect-kinesis-home.md#aws-glue-programming-etl-connect-kinesis) の「Connection types and options for ETL」(ETL の接続タイプとオプション) を参照してください。また、この接続タイプを容易にするメソッドについてのドキュメント: [create\$1data\$1frame\$1from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-create-dataframe-from-options) と、対応する Scala メソッド [def createDataFrameFromOptions](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-createDataFrameFromOptions) についてのドキュメントもご覧ください。 + `kafka` - 詳細については、AWS Glue: [Kafka 接続パラメータ](aws-glue-programming-etl-connect-kafka-home.md#aws-glue-programming-etl-connect-kafka) の「Connection types and options for ETL」(ETL の接続タイプとオプション) を参照してください。また、この接続タイプを容易にするメソッドについてのドキュメント: [create\$1data\$1frame\$1from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-create-dataframe-from-options) と、対応する Scala メソッド [def createDataFrameFromOptions](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-createDataFrameFromOptions) についてのドキュメントもご覧ください。 一部の接続タイプでは、`format_options` は必要ありません。例えば、通常の使用においては、リレーショナルデータベースへの JDBC 接続では整合性のある表形式のデータ形式でデータを取得します。したがって、JDBC 接続からの読み取りでは、`format_options` は必要ありません。 Glue でデータを読み書きする一部のメソッドでは、`format_options` が必要です。例えば、AWS Glue クローラーで `GlueContext.create_dynamic_frame.from_catalog` を使用します。クローラーによってデータの形式が決定されます。クローラーを使用する場合、AWS Glue 分類子がデータを調べ、データ形式を表す方法について賢く判断します。次に、データの表現を AWS Glue Data Catalog に保存します。これを AWS Glue ETL スクリプト内で使用し、`GlueContext.create_dynamic_frame.from_catalog` メソッドによってデータを取得できます。クローラーを使用すると、データ形式に関する情報を手動で指定する必要がなくなります。 AWS Lake Formation government table にアクセスするジョブの場合、AWS Glue は Lake Formation のgovernment table でサポートされているすべてのフォーマットの読み書きをサポートします。現在サポートされている AWS Lake Formation government table のフォーマット一覧は、「*AWS Lake Formation デベロッパーガイド*」の「[ガバメントテーブルの注意点と制限事項](https://docs.aws.amazon.com/lake-formation/latest/dg/governed-table-restrictions.html)」を参照してください。 **注記** Apache Parquet を書くために、AWS Glue ETL では、ダイナミックフレーム用に最適化されたカスタム Parquet ライタータイプのオプションを指定することで、管理されたテーブルへの書き込みのみをサポートします。`parquet` フォーマットでガバメントテーブルに書き込む場合は、テーブルパラメータの `useGlueParquetWriter` の値でキー `true` に追加する必要があります。 **Topics** + [AWS Glue でのデータ形式間の機能のサポート](#aws-glue-programming-etl-format-features) + [AWS Glue でデータ形式を操作するために使用されるパラメータ](#aws-glue-programming-etl-format-parameters) + [AWS Glue で CSV 形式を使用する](aws-glue-programming-etl-format-csv-home.md) + [AWS Glue で Parquet 形式を使用する](aws-glue-programming-etl-format-parquet-home.md) + [AWS Glue で XML 形式を使用する](aws-glue-programming-etl-format-xml-home.md) + [AWS Glue で Avro 形式を使用する](aws-glue-programming-etl-format-avro-home.md) + [AWS Glue で grokLog 形式を使用する](aws-glue-programming-etl-format-grokLog-home.md) + [AWS Glue で Ion 形式を使用する](aws-glue-programming-etl-format-ion-home.md) + [AWS Glue で JSON 形式を使用する](aws-glue-programming-etl-format-json-home.md) + [AWS Glue で ORC 形式を使用する](aws-glue-programming-etl-format-orc-home.md) + [AWS Glue ETL ジョブでのデータレイクフレームワークの使用](aws-glue-programming-etl-datalake-native-frameworks.md) + [共有設定リファレンス](#aws-glue-programming-etl-format-shared-reference) # AWS Glue で CSV 形式を使用する AWS Glue はソースからデータを取得し、さまざまなデータ形式で保存および転送されたターゲットにデータを書き込みます。このドキュメントでは、データが CSV データ形式で保存または転送される場合に、AWS Glue でデータを使用する際に利用できる機能について説明します。 AWS Glue は、カンマ区切り値 (CSV) ファイル形式をサポートしています。この形式は、最小限の行をベースにしたデータ形式です。CSV は厳密に基準に準拠していないことがよくありますが、詳細については [RFC 4180](https://tools.ietf.org/html/rfc4180) および [RFC 7111](https://tools.ietf.org/html/rfc7111) を参照してください。 AWS Glue を使用して、Amazon S3 およびストリーミングソースから CSV を読み取ることができ、Amazon S3 に CSV を書き込むこともできます。S3 から、CSV ファイルが含まれる `bzip` および `gzip` アーカイブを読み書きすることができます。このページで説明する設定ではなく、[S3 接続パラメータ](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) 上で圧縮動作を設定します。 次の表は、CSV 形式オプションをサポートする一般的な AWS Glue 機能を示しています。 | 読み込み | 書き込み | ストリーミングの読み取り | 小さなファイルのグループ化 | ジョブのブックマーク | | --- | --- | --- | --- | --- | | サポート対象 | サポート対象 | サポート対象 | サポート対象 | サポート | ## 例: S3 から CSV ファイルまたはフォルダを読み取る **前提条件:** 読み取りたい CSV ファイルまたはフォルダへの S3 パス (`s3path`) が必要です。 **設定:** 関数オプションで `format="csv"` を指定します。`connection_options` で、`paths` キーを使用して `s3path` を指定します。リーダーが S3 とやり取りする方法は、`connection_options` で設定できます。詳細については、AWS Glue: [S3 接続パラメータ](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) の「ETL の接続タイプとオプション」を参照してください。リーダーが CSV ファイルを解釈する方法は、`format_options` で設定できます。詳細については、「[CSV 設定リファレンス](#aws-glue-programming-etl-format-csv-reference)」を参照してください。 次の AWS Glue ETL スクリプトは、S3 から CSV ファイルまたはフォルダを読み取るプロセスを示しています。 `optimizePerformance` 設定キーを使用して、一般的なワークフローのパフォーマンスを最適化するカスタム CSV リーダーを提供します。このリーダーがワークロードに適しているかどうかを判断するには、「[ベクトル化された SIMD CSV リーダーで読み取りパフォーマンスを最適化する](#aws-glue-programming-etl-format-simd-csv-reader)」を参照してください。 ------ #### [ Python ] この例では [create\$1dynamic\$1frame.from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-create_dynamic_frame_from_options) メソッドを使用します。 ``` # Example: Read CSV from S3 # For show, we handle a CSV with a header row. Set the withHeader option. # Consider whether optimizePerformance is right for your workflow. from pyspark.context import SparkContext from awsglue.context import GlueContext sc = SparkContext.getOrCreate() glueContext = GlueContext(sc) spark = glueContext.spark_session dynamicFrame = glueContext.create_dynamic_frame.from_options( connection_type="s3", connection_options={"paths": ["s3://s3path"]}, format="csv", format_options={ "withHeader": True, # "optimizePerformance": True, }, ) ``` スクリプト (`pyspark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` dataFrame = spark.read\ .format("csv")\ .option("header", "true")\ .load("s3://s3path") ``` ------ #### [ Scala ] この例では [getSourceWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSourceWithFormat) 操作を使用します。 ``` // Example: Read CSV from S3 // For show, we handle a CSV with a header row. Set the withHeader option. // Consider whether optimizePerformance is right for your workflow. import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.{DynamicFrame, GlueContext} import org.apache.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) val dynamicFrame = glueContext.getSourceWithFormat( formatOptions=JsonOptions("""{"withHeader": true}"""), connectionType="s3", format="csv", options=JsonOptions("""{"paths": ["s3://s3path"], "recurse": true}""") ).getDynamicFrame() } } ``` スクリプト (`org.apache.spark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` val dataFrame = spark.read .option("header","true") .format("csv") .load("s3://s3path“) ``` ------ ## 例: CSV ファイルおよびフォルダを S3 に書き込む **前提条件:** 初期化された DataFrame (`dataFrame`) または DynamicFrame (`dynamicFrame`) が必要です。また、予想される S3 出力パスである `s3path` も必要になります。 **設定:** 関数オプションで `format="csv"` を指定します。`connection_options` で、`paths` キーを使用して `s3path` を指定します。ライターが S3 とやり取りする方法は、`connection_options` で設定できます。詳細については、AWS Glue: [S3 接続パラメータ](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) の「ETL の接続タイプとオプション」を参照してください。操作によってファイルの内容が `format_options` にどのように書き込まれるかを設定できます。詳細については、「[CSV 設定リファレンス](#aws-glue-programming-etl-format-csv-reference)」を参照してください。次の AWS Glue ETL スクリプトは、S3 に CSV ファイルとフォルダを書き込むプロセスを示しています。 ------ #### [ Python ] この例では [write\$1dynamic\$1frame.from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-write_dynamic_frame_from_options) メソッドを使用します。 ``` # Example: Write CSV to S3 # For show, customize how we write string type values. Set quoteChar to -1 so our values are not quoted. from pyspark.context import SparkContext from awsglue.context import GlueContext sc = SparkContext.getOrCreate() glueContext = GlueContext(sc) glueContext.write_dynamic_frame.from_options( frame=dynamicFrame, connection_type="s3", connection_options={"path": "s3://s3path"}, format="csv", format_options={ "quoteChar": -1, }, ) ``` スクリプト (`pyspark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` dataFrame.write\ .format("csv")\ .option("quote", None)\ .mode("append")\ .save("s3://s3path") ``` ------ #### [ Scala ] この例では [getSinkWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSinkWithFormat) メソッドを使用します。 ``` // Example: Write CSV to S3 // For show, customize how we write string type values. Set quoteChar to -1 so our values are not quoted. import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.{DynamicFrame, GlueContext} import org.apache.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) glueContext.getSinkWithFormat( connectionType="s3", options=JsonOptions("""{"path": "s3://s3path"}"""), format="csv" ).writeDynamicFrame(dynamicFrame) } } ``` スクリプト (`org.apache.spark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` dataFrame.write .format("csv") .option("quote", null) .mode("Append") .save("s3://s3path") ``` ------ ## CSV 設定リファレンス AWS Glue ライブラリが `format="csv"` を指定している場合には、以下の `format_options` を使用することができます。 + `separator` - 区切り文字を指定します。デフォルトはカンマですが、他の任意の文字を指定できます。 + **タイプ:** テキスト、**デフォルト:** `","` + `escaper` - エスケープに使用する文字を指定します。このオプションは、CSV ファイルを書き込む場合ではなく、読み取る場合にのみ使用します。有効にすると、直後の文字はそのまま使用されます。ただし、よく知られている小さいエスケープセット (`\n`、`\r`、`\t`、`\0`) を除きます。 + **タイプ:** テキスト、**デフォルト:** なし + `quoteChar` - 引用に使用する文字を指定します。デフォルト値は二重引用符です。これに `-1` を設定すると、全体的に引用が無効になります。 + **タイプ:** テキスト、**デフォルト:** `'"'` + `multiLine` - 単一のレコードが複数行にまたがることができるかどうかを指定します。これが発生するのは、フィールドに引用符で囲まれた改行文字がある場合などです。複数行にまたがるレコードがある場合は、このオプションを `True` に設定する必要があります。`multiLine` を有効にすると、解析の際にファイル分割をより慎重に行う必要があるため、パフォーマンスが低下する可能性があります。 + **タイプ:** ブール値、**デフォルト:** `false` + `withHeader` - 最初の行をヘッダーとして扱うかどうかを指定します。このオプションは `DynamicFrameReader` クラスで使用できます。 + **タイプ:** ブール値、**デフォルト:** `false` + `writeHeader` - 出力にヘッダーを書き込むかどうかを指定します。このオプションは `DynamicFrameWriter` クラスで使用できます。 + **タイプ:** ブール値、**デフォルト:** `true` + `skipFirst` - 最初のデータ行をスキップするかどうかを指定します。 + **タイプ:** ブール値、**デフォルト:** `false` + `optimizePerformance` - 高度な SIMD CSV リーダーで、Apache Arrow ベースの列指向メモリ形式を使用するかどうかを指定します。これは、AWS Glue 3.0 以降でのみ使用できます。 + **タイプ:** ブール値、**デフォルト:** `false` + `strictCheckForQuoting` – CSV を作成する際、Glue は文字列と解釈した値に引用符を追加することがあります。これは、書き出される内容があいまいにならないようにするためです。Glue は、書き出す内容を決める時間を節約するために、特定の状況では引用符が不要でも引用符を追加することがあります。厳密なチェックを有効にすると、より強力な計算が行われ、厳密に必要な場合にのみ引用符が追加されます。これは、AWS Glue 3.0 以降でのみ使用できます。 + **タイプ:** ブール値、**デフォルト:** `false` ## ベクトル化された SIMD CSV リーダーで読み取りパフォーマンスを最適化する AWS Glue バージョン 3.0 では、行ベースの CSV リーダーと比較して、全体的なジョブパフォーマンスを大幅に高速化できる最適化された CSV リーダーが追加されています。 最適化されたリーダーでは: + CPU SIMD 命令を使用してディスクから読み取ります + レコードを列形式 (Apache Arrow) で即座にメモリに書き込みます + レコードをバッチに分割します これにより、レコードがバッチ処理されるときや、後で列形式に変換されるときの処理時間が節約されます。一例としては、スキーマを変更したり、列ごとにデータを取得したりする場合などが挙げられます。 最適化されたリーダーを使用するには、`format_options` またはテーブルプロパティで `"optimizePerformance"` を `true` に設定します。 ``` glueContext.create_dynamic_frame.from_options( frame = datasource1, connection_type = "s3", connection_options = {"paths": ["s3://s3path"]}, format = "csv", format_options={ "optimizePerformance": True, "separator": "," }, transformation_ctx = "datasink2") ``` **ベクトル化された CSV リーダーでの制限事項** ベクトル化された CSV リーダーには、次の制限があるので注意が必要です。 + `multiLine` および `escaper` 形式オプションはサポートされません。デフォルトの `escaper` として二重引用符文字 `'"'` を使用します。これらのオプションを設定すると、AWS Glue は自動的にフォールバックし、行ベースの CSV リーダーを使用するようになります。 + [ChoiceType](https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-crawler-pyspark-extensions-types.html#aws-glue-api-crawler-pyspark-extensions-types-awsglue-choicetype) の DynamicFrame の作成はサポートされません。 + [error records](https://docs.aws.amazon.com/glue/latest/dg/glue-etl-scala-apis-glue-dynamicframe-class.html#glue-etl-scala-apis-glue-dynamicframe-class-defs-errorsAsDynamicFrame) の DynamicFrame の作成はサポートされません。 + 日本語や中国語など、マルチバイト文字を含む CSV ファイルからの読み取りはサポートされません。 # AWS Glue で Parquet 形式を使用する AWS Glue はソースからデータを取得し、さまざまなデータ形式で保存および転送されたターゲットにデータを書き込みます。このドキュメントでは、データが Parquet データ形式で保存または転送される場合に、AWS Glue でデータを使用する際に利用できる機能について説明します。 AWS Glue は Parquet 形式の使用をサポートしています。この形式は、パフォーマンス指向の列ベースのデータ形式です。標準局から発行されている形式の概要については、「[Apache Parquet ドキュメントの概要](https://parquet.apache.org/docs/overview/)」を参照してください。 AWS Glue を使用して、Amazon S3 およびストリーミングソースから Parquet ファイルを読み取ることができ、Amazon S3 に Parquet ファイルを書き込むこともできます。S3 から、Parquet ファイルが含まれる `bzip` および `gzip` アーカイブを読み書きすることができます。このページで説明する設定ではなく、[S3 接続パラメータ](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) 上で圧縮動作を設定します。 次の表は、Parquet 形式オプションをサポートする一般的な AWS Glue 機能を示しています。 | 読み込み | 書き込み | ストリーミングの読み取り | 小さなファイルのグループ化 | ジョブのブックマーク | | --- | --- | --- | --- | --- | | サポート対象 | サポート対象 | サポート | サポートされていません | サポート対象\$1 | \$1 AWS Glue バージョン 1.0\$1 でサポート ## 例: S3 から Parquet ファイルまたはフォルダを読み取る **前提条件:** 読み取りたい Parquet ファイルまたはフォルダへの S3 パス (`s3path`) が必要です。 **設定:** 関数オプションで `format="parquet"` を指定します。`connection_options` で、`paths` キーを使用して `s3path` を指定します。 リーダーが S3 とやり取りする方法は、`connection_options` で設定できます。詳細については、AWS Glue: [S3 接続パラメータ](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) の「ETL の接続タイプとオプション」を参照してください。 リーダーが Parquet ファイルを解釈する方法は、`format_options` で設定できます。詳細については、「[Parquet 設定リファレンス](#aws-glue-programming-etl-format-parquet-reference)」を参照してください。 次の AWS Glue ETL スクリプトは、S3 から Parquet ファイルまたはフォルダを読み取るプロセスを示しています。 ------ #### [ Python ] この例では [create\$1dynamic\$1frame.from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-create_dynamic_frame_from_options) メソッドを使用します。 ``` # Example: Read Parquet from S3 from pyspark.context import SparkContext from awsglue.context import GlueContext sc = SparkContext.getOrCreate() glueContext = GlueContext(sc) spark = glueContext.spark_session dynamicFrame = glueContext.create_dynamic_frame.from_options( connection_type = "s3", connection_options = {"paths": ["s3://s3path/"]}, format = "parquet" ) ``` スクリプト (`pyspark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` dataFrame = spark.read.parquet("s3://s3path/") ``` ------ #### [ Scala ] この例では [getSourceWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSourceWithFormat) メソッドを使用します。 ``` // Example: Read Parquet from S3 import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.{DynamicFrame, GlueContext} import org.apache.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) val dynamicFrame = glueContext.getSourceWithFormat( connectionType="s3", format="parquet", options=JsonOptions("""{"paths": ["s3://s3path"]}""") ).getDynamicFrame() } } ``` スクリプト (`org.apache.spark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` spark.read.parquet("s3://s3path/") ``` ------ ## 例: Parquet ファイルおよびフォルダを S3 に書き込む **前提条件:** 初期化された DataFrame (`dataFrame`) または DynamicFrame (`dynamicFrame`) が必要です。また、予想される S3 出力パスである `s3path` も必要になります。 **設定:** 関数オプションで `format="parquet"` を指定します。`connection_options` で、`paths` キーを使用して `s3path` を指定します。 ライターが S3 と対話する方法を、`connection_options` でさらに詳しく変更することができます。詳細については、AWS Glue: [S3 接続パラメータ](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) の「ETL の接続タイプとオプション」を参照してください。操作によってファイルの内容が `format_options` にどのように書き込まれるかを設定できます。詳細については、「[Parquet 設定リファレンス](#aws-glue-programming-etl-format-parquet-reference)」を参照してください。 次の AWS Glue ETL スクリプトは、S3 に Parquet ファイルとフォルダを書き込むプロセスを示しています。 `useGlueParquetWriter` 設定キーを通して、DynamicFrame のパフォーマンスを最適化するカスタム Parquet ライターを提供します。このライターがワークロードに適しているかどうかを判断するには、「[Glue Parquet Writer](#aws-glue-programming-etl-format-glue-parquet-writer)」を参照してください。 ------ #### [ Python ] この例では [write\$1dynamic\$1frame.from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-write_dynamic_frame_from_options) メソッドを使用します。 ``` # Example: Write Parquet to S3 # Consider whether useGlueParquetWriter is right for your workflow. from pyspark.context import SparkContext from awsglue.context import GlueContext sc = SparkContext.getOrCreate() glueContext = GlueContext(sc) glueContext.write_dynamic_frame.from_options( frame=dynamicFrame, connection_type="s3", format="parquet", connection_options={ "path": "s3://s3path", }, format_options={ # "useGlueParquetWriter": True, }, ) ``` スクリプト (`pyspark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` df.write.parquet("s3://s3path/") ``` ------ #### [ Scala ] この例では [getSinkWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSinkWithFormat) メソッドを使用します。 ``` // Example: Write Parquet to S3 // Consider whether useGlueParquetWriter is right for your workflow. import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.{DynamicFrame, GlueContext} import org.apache.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) glueContext.getSinkWithFormat( connectionType="s3", options=JsonOptions("""{"path": "s3://s3path"}"""), format="parquet" ).writeDynamicFrame(dynamicFrame) } } ``` スクリプト (`org.apache.spark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` df.write.parquet("s3://s3path/") ``` ------ ## Parquet 設定リファレンス AWS Glue ライブラリが `format="parquet"` を指定している場合には、以下の `format_options` を使用することができます。 + `useGlueParquetWriter` - DynamicFrame ワークフローのパフォーマンスを最適化するカスタム Parquet ライターの使用を指定します。使用方法の詳細については、「[Glue Parquet Writer](#aws-glue-programming-etl-format-glue-parquet-writer)」を参照してください。 + **タイプ:** ブール値、**デフォルト:** `false` + `compression` - 使用する圧縮コーデックを指定します。値は `org.apache.parquet.hadoop.metadata.CompressionCodecName` との完全な互換性があります。 + **タイプ:** 列挙型テキスト、**デフォルト:** `"snappy"` + 値: `"uncompressed"`、`"snappy"`、`"gzip"`、`"lzo"` + `blockSize` - メモリにバッファされる行グループのサイズを、バイト数で指定します。これはパフォーマンスのチューニングに使用します。サイズは正確に、メガバイト数で分割する必要があります。 + **タイプ:** 数値、**デフォルト:** `134217728` + デフォルト値は 128 MB です。 + `pageSize` - ページのサイズをバイト数で指定します。これはパフォーマンスのチューニングに使用します。ページは、単一のレコードにアクセスするために完全に読み取らなければならない最小単位です。 + **タイプ:** 数値、**デフォルト:** `1048576` + デフォルト値は 1 MB です。 **注記** 加えて、基盤となる SparkSQL コードで受け入れられるすべてのオプションは、`connection_options` マップパラメータを介し、この形式に渡されます。例えば、[mergeSchema](https://spark.apache.org/docs/latest/sql-data-sources-parquet.html#schema-merging) などの Spark 設定を AWS Glue Spark リーダーに行うことで、すべてのファイルのスキーマをマージすることができます。 ## 書き込みパフォーマンスを AWS Glue Parquet ライターで最適化する **注記** AWS Glue Parquet ライターには、これまで `glueparquet` 形式タイプでアクセスしていました。このアクセス方法は非推奨となりました。代わりに、`useGlueParquetWriter` を有効にして `parquet` タイプを使用してください。 AWS Glue Parquet ライターには、Parquet ファイルの書き込みを高速化できるパフォーマンス強化が組み込まれています。これまでのライターでは、書き込む前にスキーマを計算していました。Parquet 形式では、すぐに取得できる形でスキーマを保存しないため、時間がかかることがありました。AWS Glue Parquet ライターを使用すれば、事前に計算済みのスキーマは必要なくなります。データが到着次第、ライターがスキーマを動的に計算して変更します。 `useGlueParquetWriter` を指定する場合、以下の制限に注意してください。 + ライターは、列の追加や削除などのスキーマの進化のみをサポートし、`ResolveChoice` のように列タイプの変更はサポートしません。 + ライターは、空の DataFrame の書き込み (スキーマのみのファイルを書き込む場合など) をサポートしていません。`enableUpdateCatalog=True` の設定によって AWS Glue データカタログと統合する場合、空の DataFrame を書き込もうとしてもデータカタログは更新されません。これにより、スキーマなしでデータカタログにテーブルが作成されます。 変換時にこれらの制限が必要とならない場合、AWS Glue Parquet ライターをオンにすることで、パフォーマンスが向上するはずです。 # AWS Glue で XML 形式を使用する AWS Glue はソースからデータを取得し、さまざまなデータ形式で保存および転送されたターゲットにデータを書き込みます。このドキュメントでは、データが XML データ形式で保存または転送される場合に、AWS Glue でデータを使用する際に利用できる機能について説明します。 AWS Glue は XML 形式の使用をサポートしています。この形式は、行ベースでも列ベースでもない、非常に設定可能性が高く、厳密に定義されたデータ構造を表します。XML は高度に標準化されています。標準局から発行されている形式の概要については、「[XML エッセンシャル](https://www.w3.org/standards/xml/core)」を参照してください。 AWS Glue を使用して、Amazon S3 から XML ファイルを読み込んだり、XML ファイルが含まれる `bzip` および `gzip` アーカイブを読み込んだりすることができます。このページで説明する設定ではなく、[S3 接続パラメータ](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) 上で圧縮動作を設定します。 次の表は、XML 形式オプションをサポートする一般的な AWS Glue 機能を示しています。 | 読み込み | 書き込み | ストリーミングの読み取り | 小さなファイルのグループ化 | ジョブのブックマーク | | --- | --- | --- | --- | --- | | サポート | サポートされていません | サポートされていません | サポート | サポート | ## 例: S3 から XML を読み込む XML リーダーは XML タグ名を取得します。入力内にあるそのタグを持つ要素を調べてスキーマを推測し、対応する値を DynamicFrame に入力します。AWS Glue XML の機能は、[Apache Spark の XML データソース](https://github.com/databricks/spark-xml)に類似した動作をします。このリーダーをそのプロジェクトのドキュメントと比較することで、基本的な動作に関する洞察が得られる可能性があります。 **前提条件:** 読み取りたい XML ファイルまたはフォルダへの S3 パス (`s3path`) と、XML ファイルに関するいくつかの情報が必要です。また、読み込みたい XML 要素のタグである `xmlTag` が必要です。 **設定:** 関数オプションで `format="xml"` を指定します。`connection_options` で、`paths` キーを使用して `s3path` を指定します。リーダーが S3 とやり取りする方法は、`connection_options` でさらに詳しく設定できます。詳細については、AWS Glue: [S3 接続パラメータ](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) の「ETL の接続タイプとオプション」を参照してください。`format_options` で、`rowTag` キーを使用して `xmlTag` を指定します。リーダーが XML ファイルを解釈する方法は、`format_options` でさらに詳しく設定できます。詳細については、「[XML 設定リファレンス](#aws-glue-programming-etl-format-xml-reference)」を参照してください。 次の AWS Glue ETL スクリプトは、S3 から XML ファイルまたはフォルダを読み取るプロセスを示しています。 ------ #### [ Python ] この例では [create\$1dynamic\$1frame.from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-create_dynamic_frame_from_options) メソッドを使用します。 ``` # Example: Read XML from S3 # Set the rowTag option to configure the reader. from awsglue.context import GlueContext from pyspark.context import SparkContext sc = SparkContext.getOrCreate() glueContext = GlueContext(sc) dynamicFrame = glueContext.create_dynamic_frame.from_options( connection_type="s3", connection_options={"paths": ["s3://s3path"]}, format="xml", format_options={"rowTag": "xmlTag"}, ) ``` スクリプト (`pyspark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` dataFrame = spark.read\ .format("xml")\ .option("rowTag", "xmlTag")\ .load("s3://s3path") ``` ------ #### [ Scala ] この例では [getSourceWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSourceWithFormat) 操作を使用します。 ``` // Example: Read XML from S3 // Set the rowTag option to configure the reader. import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.GlueContext import org.apache.spark.sql.SparkSession val glueContext = new GlueContext(SparkContext.getOrCreate()) val sparkSession: SparkSession = glueContext.getSparkSession object GlueApp { def main(sysArgs: Array[String]): Unit = { val dynamicFrame = glueContext.getSourceWithFormat( formatOptions=JsonOptions("""{"rowTag": "xmlTag"}"""), connectionType="s3", format="xml", options=JsonOptions("""{"paths": ["s3://s3path"], "recurse": true}""") ).getDynamicFrame() } ``` スクリプト (`org.apache.spark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` val dataFrame = spark.read .option("rowTag", "xmlTag") .format("xml") .load("s3://s3path“) ``` ------ ## XML 設定リファレンス AWS Glue ライブラリが `format="xml"` を指定している場合には、以下の `format_options` を使用することができます。 + `rowTag` - 行として扱うファイル内の XML タグを指定します。行のタグを自己終了型にすることはできません。 + **タイプ:** テキスト、**必須** + `encoding` - 文字エンコードを指定します。これは、ランタイム環境がサポートする [Charset](https://docs.oracle.com/javase/8/docs/api/java/nio/charset/Charset.html) の名前またはエイリアスにすることができます。エンコーディングのサポートに関しては特に保証していませんが、主要なエンコーディングは動作するはずです。 + **タイプ:** テキスト、**デフォルト:** `"UTF-8"` + `excludeAttribute` - 要素の属性を除外するかどうかを指定します。 + **タイプ:** ブール値、**デフォルト:** `false` + `treatEmptyValuesAsNulls` - 空白文字を null 値として扱うかどうかをします。 + **タイプ:** ブール値、**デフォルト:** `false` + `attributePrefix` - 子要素テキストから区別するために属性に付加するプレフィックス。このプレフィックスをフィールド名として使用します。 + **タイプ:** テキスト、**デフォルト:** `"_"` + `valueTag` - 要素内に子を持たない属性がある場合、値に使用するタグ。 + **タイプ:** テキスト、**デフォルト:** `"_VALUE"` + `ignoreSurroundingSpaces` - 値を囲む空白文字を無視するかどうかを指定します。 + **タイプ:** ブール値、**デフォルト:** `false` + `withSchema` - 推論されたスキーマを上書きしたい場合に、期待されるスキーマを含みます。このオプションを使用しない場合は、AWS Glue で XML データからスキーマを推定します。 + **タイプ:** テキスト、**デフォルト:** 該当なし + 値は `StructType` を表す JSON オブジェクトである必要があります。 ## XML スキーマを手動で指定する **手動 XML スキーマの例** これは、XML データ用にスキーマを指定するために、`withSchema` 形式オプションを使用する場合の例です。 ``` from awsglue.gluetypes import * schema = StructType([ Field("id", IntegerType()), Field("name", StringType()), Field("nested", StructType([ Field("x", IntegerType()), Field("y", StringType()), Field("z", ChoiceType([IntegerType(), StringType()])) ])) ]) datasource0 = create_dynamic_frame_from_options( connection_type, connection_options={"paths": ["s3://xml_bucket/someprefix"]}, format="xml", format_options={"withSchema": json.dumps(schema.jsonValue())}, transformation_ctx = "" ) ``` # AWS Glue で Avro 形式を使用する AWS Glue はソースからデータを取得し、さまざまなデータ形式で保存および転送されたターゲットにデータを書き込みます。このドキュメントでは、データが Avro データ形式で保存または転送される場合に、AWS Glue でデータを使用する際に利用できる機能について説明します。 AWS Glue は Avro 形式の使用をサポートしています。この形式は、パフォーマンス指向の行ベースのデータ形式です。標準局による形式の概要については、「[Apache Avro 1.8.2 Documentation](https://avro.apache.org/docs/1.8.2/)」(Apache Avro ドキュメント 1.8.2) を参照してください。 AWS Glue を使用して、Amazon S3 およびストリーミングソースから Avro ファイルを読み取り、Avro ファイルを Amazon S3 に書き込むことができます。S3 から、Avro ファイルが含まれる `bzip2` および `gzip` アーカイブを読み書きすることができます。さらに、Avro ファイルが含まれる `deflate`、`snappy`、`xz` アーカイブを書き込むこともできます。このページで説明する設定ではなく、[S3 接続パラメータ](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) 上で圧縮動作を設定します。 次の表は、Avro 形式のオプションをサポートする一般的な AWS Glue オペレーションを示しています。 | 読み込み | 書き込み | ストリーミングの読み取り | 小さなファイルのグループ化 | ジョブのブックマーク | | --- | --- | --- | --- | --- | | サポート | サポート | サポート対象\$1 | サポートされていません | サポート | 制限付きで、\$1 がサポートされています。詳細については、「[Avro ストリーミングソースに関する注意事項と制約事項](add-job-streaming.md#streaming-avro-notes)」を参照してください。 ## 例: S3 から Avro ファイルまたはフォルダを読み取る **前提条件:** 読み取る Avro ファイルまたはフォルダへの S3 パス (`s3path`) が必要です。 **設定:** 関数オプションで `format="avro"` を指定します。`connection_options` で、`paths` キーを使用して `s3path` を指定します。リーダーが S3 とやり取りする方法は、`connection_options` で設定できます。詳細については、AWS Glue: [Amazon S3 接続のオプションのリファレンス](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) の「ETL の入力および出力のデータ形式オプション」を参照してください。リーダーが Avro ファイルを解釈する方法は、`format_options` で設定できます。詳細については、[Avro 設定リファレンス](#aws-glue-programming-etl-format-avro-reference)を参照してください。 次の AWS Glue ETL スクリプトは、S3 から Avro ファイルまたはフォルダを読み取るプロセスを示しています。 ------ #### [ Python ] この例では [create\$1dynamic\$1frame.from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-create_dynamic_frame_from_options) メソッドを使用します。 ``` from pyspark.context import SparkContext from awsglue.context import GlueContext sc = SparkContext.getOrCreate() glueContext = GlueContext(sc) dynamicFrame = glueContext.create_dynamic_frame.from_options( connection_type="s3", connection_options={"paths": ["s3://s3path"]}, format="avro" ) ``` ------ #### [ Scala ] この例では [getSourceWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSourceWithFormat) 操作を使用します。 ``` import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.GlueContext import org.apache.spark.sql.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) val dynamicFrame = glueContext.getSourceWithFormat( connectionType="s3", format="avro", options=JsonOptions("""{"paths": ["s3://s3path"]}""") ).getDynamicFrame() } ``` ------ ## 例: Avro ファイルおよびフォルダを S3 に書き込む **前提条件:** 初期化された DataFrame (`dataFrame`) または DynamicFrame (`dynamicFrame`) が必要です。また、予想される S3 出力パスである `s3path` も必要になります。 **設定:** 関数オプションで `format="avro"` を指定します。`connection_options` で、`paths` キーを使用して `s3path` を指定します。ライターが S3 と対話する方法を、`connection_options` でさらに詳しく変更することができます。詳細については、AWS Glue: [Amazon S3 接続のオプションのリファレンス](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) の「ETL の入力および出力のデータ形式オプション」を参照してください。リーダーが Avro ファイルを解釈する方法は、`format_options` で変更できます。詳細については、[Avro 設定リファレンス](#aws-glue-programming-etl-format-avro-reference)を参照してください。 次の AWS Glue ETL スクリプトは、S3 に Avro ファイルまたはフォルダを書き込むプロセスを示しています。 ------ #### [ Python ] この例では [write\$1dynamic\$1frame.from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-write_dynamic_frame_from_options) メソッドを使用します。 ``` from pyspark.context import SparkContext from awsglue.context import GlueContext sc = SparkContext.getOrCreate() glueContext = GlueContext(sc) glueContext.write_dynamic_frame.from_options( frame=dynamicFrame, connection_type="s3", format="avro", connection_options={ "path": "s3://s3path" } ) ``` ------ #### [ Scala ] この例では [getSinkWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSinkWithFormat) メソッドを使用します。 ``` import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.{DynamicFrame, GlueContext} import org.apache.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) glueContext.getSinkWithFormat( connectionType="s3", options=JsonOptions("""{"path": "s3://s3path"}"""), format="avro" ).writeDynamicFrame(dynamicFrame) } } ``` ------ ## Avro 設定リファレンス AWS Glue ライブラリが `format="avro"` を指定する場合は、以下の `format_options` を使用することができます。 + `version` – サポートする Apache Avro リーダー/ライター形式のバージョンを指定します。デフォルト値は「1.7」です。Avro 論理型の読み取りと書き込みを有効にするには、`format_options={"version": “1.8”}` を指定します。詳細については、「[Apache Avro 1.7.7 の仕様](https://avro.apache.org/docs/1.7.7/spec.html)」および「[Apache Avro 1.8.2 の仕様](https://avro.apache.org/docs/1.8.2/spec.html)」を参照してください。 Apache Avro 1.8 コネクタは、次の論理型変換をサポートしています。 リーダーの場合: この表に、Avro リーダー 1.7 と 1.8 について、Avro データ型 (論理型と Avro プリミティブ型) と AWS Glue`DynamicFrame` データ型の間での型変換を示します。 | Avro データ型: 論理型 | Avro データ型: Avro プリミティブ型 | GlueDynamicFrame データ型:Avro リーダー 1.7 | GlueDynamicFrame データ型: Avro リーダー 1.8 | | --- | --- | --- | --- | | 10 進数 | bytes | BINARY | 10 進数 | | 10 進数 | 固定 | BINARY | 10 進数 | | 日付 | int | INT | 日付 | | 時間 (ミリ秒) | int | INT | INT | | 時間 (マイクロ秒) | long | LONG | LONG | | タイムスタンプ (ミリ秒) | long | LONG | Timestamp | | タイムスタンプ (マイクロ秒) | long | LONG | LONG | | 所要時間 (論理型ではない) | 12 で固定 | BINARY | BINARY | ライターの場合: この表に、Avro ライター 1.7 と 1.8 について、データ型と Avro データ型 AWS Glue`DynamicFrame` の間での型変換を示します。 | AWS Glue`DynamicFrame` データ型 | Avro データ型:Avro ライター 1.7 | Avro データ型:Avro タイター 1.8 | | --- | --- | --- | | 10 進数 | 文字列 | decimal | | 日付 | 文字列 | date | | Timestamp | 文字列 | timestamp-micros | ## Avro Spark DataFrame サポート Spark DataFrame API から Avro を使用するには、対応する Spark のバージョン用の Spark Avro プラグインをインストールする必要があります。ジョブで使用できる Spark のバージョンは、AWS Glue のバージョンによって決まります。Spark のバージョンの詳細については、「[AWS Glue バージョン](release-notes.md)」を参照してください。このプラグインは Apache によって管理されており、特定のサポートを保証するものではありません。 AWS Glue 2.0 - Spark Avro プラグインのバージョン 2.4.3 を使用します。この JAR は Maven Central にあります。[org.apache.spark:spark-avro\$12.12:2.4.3](https://search.maven.org/artifact/org.apache.spark/spark-avro_2.12/3.1.1/jar) を参照してください。 AWS Glue 3.0 - Spark Avro プラグインのバージョン 3.1.1 を使用します。この JAR は Maven Central にあります。[org.apache.spark:spark-avro\$12.12:3.1.1](https://search.maven.org/artifact/org.apache.spark/spark-avro_2.12/3.1.1/jar) を参照してください。 AWS Glue の ETL ジョブに追加の JAR を含めるには、`--extra-jars` ジョブパラメータを使用します。ジョブパラメータについては、「[AWS Glue ジョブでジョブパラメータを使用する](aws-glue-programming-etl-glue-arguments.md)」を参照してください。AWS マネジメントコンソール で、このパラメータを設定することもできます。 # AWS Glue で grokLog 形式を使用する AWS Glue はソースからデータを取得し、さまざまなデータ形式で保存および転送されたターゲットにデータを書き込みます。このドキュメントでは、データが大まかに構造化されたプレーンテキスト形式で保存または転送される場合に、AWS Glue でデータを使用する際に利用できる機能を紹介します。 AWS Glue は、Grok パターンの使用をサポートしています。Grok パターンは、正規表現のキャプチャグループに似ています。プレーンテキストファイル内の文字シーケンスのパターンを認識し、タイプと目的を与えます。AWS Glue での主な目的はログを読み取ることです。作成者による Grokの概要については、「[Logstash Reference: Grok filter plugin](https://www.elastic.co/guide/en/logstash/current/plugins-filters-grok.html)」(Logstash リファレンス: Grok フィルタープラグイン) を参照してください。 | 読み込み | 書き込み | ストリーミングの読み取り | 小さなファイルのグループ化 | ジョブのブックマーク | | --- | --- | --- | --- | --- | | サポート | 該当しません | サポート | サポート | サポートされていません | ## grokLog 設定リファレンス `format="grokLog"` には、以下の `format_options` 値を使用できます。 + `logFormat` – ログの形式と一致する Grok パターンを指定します。 + `customPatterns` – ここで使用する追加の Grok パターンを指定します。 + `MISSING` – 欠落した値の識別に使用するシグナルを指定します。デフォルトは `'-'` です。 + `LineCount` – 各ログレコードの行数を指定します。デフォルト値は `'1'` です。現在 1 行のレコードのみがサポートされています。 + `StrictMode` – strict モードを有効にするかどうかを指定するブール値。厳格モードでは、リーダーは自動的な型変換や復旧を行いません。デフォルト値は `"false"` です。 # AWS Glue で Ion 形式を使用する AWS Glue はソースからデータを取得し、さまざまなデータ形式で保存および転送されたターゲットにデータを書き込みます。このドキュメントでは、データが Ion データ形式で保存または転送される場合に、AWS Glue でデータを使用する際に利用できる機能を紹介します。 AWS Glue は Ion 形式の使用をサポートしています。この形式は、データ構造 (行または列ベースではない) を交換可能なバイナリおよびプレーンテキスト形式で表します。作成者による形式の概要については、「[Amazon Ion](https://amzn.github.io/ion-docs/)」を参照してください。詳細については、[Amazon Ion の仕様](https://amzn.github.io/ion-docs/spec.html)に関するドキュメントを参照してください。 AWS Glue を使用して、Amazon S3 から Ion ファイルを読み取ることができます。S3 から、Ion ファイルを含む `bzip` および `gzip` アーカイブを読み取ることができます。このページで説明する設定ではなく、[S3 接続パラメータ](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) 上で圧縮動作を設定します。 次の表は、Ion 形式のオプションをサポートする一般的な AWS Glue オペレーションを示しています。 | 読み込み | 書き込み | ストリーミングの読み取り | 小さなファイルのグループ化 | ジョブのブックマーク | | --- | --- | --- | --- | --- | | サポート | サポートされていません | サポートされていません | サポート | サポートされていません | ## 例: S3 から Ion ファイルとフォルダを読み取る **前提条件:** 読み取る Ion ファイルまたはフォルダへの S3 パス (`s3path`) が必要です。 **設定:** 関数オプションで `format="json"` を指定します。`connection_options` で、`paths` キーを使用して `s3path` を指定します。リーダーが S3 とやり取りする方法は、`connection_options` で設定できます。詳細については、AWS Glue: [Amazon S3 接続のオプションのリファレンス](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) の「ETL の接続タイプとオプション」を参照してください。 次の AWS Glue ETL スクリプトは、S3 から Ion ファイルまたはフォルダを読み取るプロセスを示しています。 ------ #### [ Python ] この例では [create\$1dynamic\$1frame.from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-create_dynamic_frame_from_options) メソッドを使用します。 ``` # Example: Read ION from S3 from pyspark.context import SparkContext from awsglue.context import GlueContext sc = SparkContext.getOrCreate() glueContext = GlueContext(sc) dynamicFrame = glueContext.create_dynamic_frame.from_options( connection_type="s3", connection_options={"paths": ["s3://s3path"]}, format="ion" ) ``` ------ #### [ Scala ] この例では [getSourceWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSourceWithFormat) 操作を使用します。 ``` // Example: Read ION from S3 import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.GlueContext import org.apache.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) val dynamicFrame = glueContext.getSourceWithFormat( connectionType="s3", format="ion", options=JsonOptions("""{"paths": ["s3://s3path"], "recurse": true}""") ).getDynamicFrame() } } ``` ------ ## Ion 設定リファレンス `format="ion"` の `format_options` 値はありません。 # AWS Glue で JSON 形式を使用する AWS Glue はソースからデータを取得し、さまざまなデータ形式で保存および転送されたターゲットにデータを書き込みます。このドキュメントでは、JSON データ形式で保存または転送されているデータを、AWS Glue で使用するために用意されている機能について説明します。 AWS Glue は、JSON 形式の使用をサポートしています。この形式は、行ベースまたは列ベースではなく、形状は一貫しているが内容は柔軟なデータ構造を表します。JSON は、複数の機関が発行する並列した規格で定義されており、そのうちの 1 つが ECMA-404 です。一般的に参照されるソースによる形式の概要については、「[Introducing JSON](https://www.json.org/)」(JSON の概要) を参照してください。 AWS Glue を使用して、JSON ファイルや `bzip` および `gzip` などの圧縮された JSON ファイルを Amazon S3 から 読み取ることができます。このページで説明する設定ではなく、[S3 接続パラメータ](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) 上で圧縮動作を設定します。 | 読み込み | 書き込み | ストリーミングの読み取り | 小さなファイルのグループ化 | ジョブのブックマーク | | --- | --- | --- | --- | --- | | サポート | サポート対象 | サポート対象 | サポート対象 | サポート | ## 例: S3 から JSON ファイルまたはフォルダを読み取る **前提条件:** 読み取る JSON ファイルまたはフォルダへの S3 パス (`s3path`) が必要です。 **設定:** 関数オプションで `format="json"` を指定します。`connection_options` で、`paths` キーを使用して `s3path` を指定します。さらに、接続オプションで、読み取り操作がS3 を通過する方法を変更することができます。詳細については、「[Amazon S3 接続のオプションのリファレンス](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3)」を参照してください。リーダーが JSON ファイルを解釈する方法は、`format_options` で設定できます。詳細については、[JSON 設定リファレンス](#aws-glue-programming-etl-format-json-reference)を参照してください。 次の AWS Glue ETL スクリプトは、S3 から JSON ファイルまたはフォルダを読み取るプロセスを示しています。 ------ #### [ Python ] この例では [create\$1dynamic\$1frame.from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-create_dynamic_frame_from_options) メソッドを使用します。 ``` # Example: Read JSON from S3 # For show, we handle a nested JSON file that we can limit with the JsonPath parameter # For show, we also handle a JSON where a single entry spans multiple lines # Consider whether optimizePerformance is right for your workflow. from pyspark.context import SparkContext from awsglue.context import GlueContext sc = SparkContext.getOrCreate() glueContext = GlueContext(sc) spark = glueContext.spark_session dynamicFrame = glueContext.create_dynamic_frame.from_options( connection_type="s3", connection_options={"paths": ["s3://s3path"]}, format="json", format_options={ "jsonPath": "$.id", "multiline": True, # "optimizePerformance": True, -> not compatible with jsonPath, multiline } ) ``` スクリプト (`pyspark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` dataFrame = spark.read\ .option("multiline", "true")\ .json("s3://s3path") ``` ------ #### [ Scala ] この例では [getSourceWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSourceWithFormat) 操作を使用します。 ``` // Example: Read JSON from S3 // For show, we handle a nested JSON file that we can limit with the JsonPath parameter // For show, we also handle a JSON where a single entry spans multiple lines // Consider whether optimizePerformance is right for your workflow. import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.{DynamicFrame, GlueContext} import org.apache.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) val dynamicFrame = glueContext.getSourceWithFormat( formatOptions=JsonOptions("""{"jsonPath": "$.id", "multiline": true, "optimizePerformance":false}"""), connectionType="s3", format="json", options=JsonOptions("""{"paths": ["s3://s3path"], "recurse": true}""") ).getDynamicFrame() } } ``` スクリプト (`pyspark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` val dataFrame = spark.read .option("multiline", "true") .json("s3://s3path") ``` ------ ## 例: JSON ファイルおよびフォルダを S3 に書き込む **前提条件:** 初期化された DataFrame (`dataFrame`) または DynamicFrame (`dynamicFrame`) が必要です。また、予想される S3 出力パスである `s3path` も必要になります。 **設定:** 関数オプションで `format="json"` を指定します。`connection_options` で、`paths` キーを使用して `s3path` を指定します。ライターが S3 と対話する方法を、`connection_options` でさらに詳しく変更することができます。詳細については、AWS Glue: [Amazon S3 接続のオプションのリファレンス](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) の「ETL の入力および出力のデータ形式オプション」を参照してください。ライターが JSON ファイルを解釈する方法は、`format_options` で設定できます。詳細については、[JSON 設定リファレンス](#aws-glue-programming-etl-format-json-reference)を参照してください。 次の AWS Glue ETL スクリプトは、S3 から JSON ファイルまたはフォルダを書き込むプロセスを示しています。 ------ #### [ Python ] この例では [write\$1dynamic\$1frame.from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-write_dynamic_frame_from_options) メソッドを使用します。 ``` # Example: Write JSON to S3 from pyspark.context import SparkContext from awsglue.context import GlueContext sc = SparkContext.getOrCreate() glueContext = GlueContext(sc) glueContext.write_dynamic_frame.from_options( frame=dynamicFrame, connection_type="s3", connection_options={"path": "s3://s3path"}, format="json" ) ``` スクリプト (`pyspark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` df.write.json("s3://s3path/") ``` ------ #### [ Scala ] この例では [getSinkWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSinkWithFormat) メソッドを使用します。 ``` // Example: Write JSON to S3 import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.{DynamicFrame, GlueContext} import org.apache.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) glueContext.getSinkWithFormat( connectionType="s3", options=JsonOptions("""{"path": "s3://s3path"}"""), format="json" ).writeDynamicFrame(dynamicFrame) } } ``` スクリプト (`pyspark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` df.write.json("s3://s3path") ``` ------ ## JSON 設定リファレンス `format="json"` には、以下の `format_options` 値を使用できます。 + `jsonPath` – レコードに読み込むオブジェクトを特定する [JsonPath](https://github.com/json-path/JsonPath) 式。これは、ファイルが外部配列内にネストされたレコードを含む場合に役立ちます。たとえば、次の JsonPath 式のターゲットは、JSON オブジェクトの `id` フィールドです。 ``` format="json", format_options={"jsonPath": "$.id"} ``` + `multiline` — 単一のレコードが複数行にまたがることができるかどうかを指定するブール値。これが発生するのは、フィールドに引用符で囲まれた改行文字がある場合などです。複数行にまたがるレコードがある場合は、このオプションを `"true"` に設定する必要があります。デフォルト値は `"false"` であり、解析時によりアグレッシブなファイル分割を可能にします。 + `optimizePerformance` – 高度な SIMD JSON リーダーで、Apache Arrow ベースの列指向メモリ形式を使用するかどうかを指定するブール値。これは、AWS Glue 3.0 でのみ使用できます。`multiline` または `jsonPath` と互換性がありません。これらのオプションのいずれかを指定すると、AWS Glue は標準のリーダーにフォールバックするように指示されます。 + `withSchema` — 「[XML スキーマを手動で指定する](aws-glue-programming-etl-format-xml-home.md#aws-glue-programming-etl-format-xml-withschema)」で説明されている形式でテーブルスキーマを指定する文字列の値。非カタログ接続から読み取る場合にのみ、`optimizePerformance` で使用されます。 ## Apache Arrow 列指向形式によりベクトル化された SIMD JSON リーダーの使用 AWS Glue バージョン 3.0 では、JSON データ用のベクター化されたリーダーが追加されています。特定の条件下では、標準のリーダーと比較して 2 倍高速に動作します。このリーダーには、このセクションに記載されているように、ユーザーが使用前に知っておく必要のある特定の制限があります。 最適化されたリーダーを使用するには、`format_options` またはテーブルプロパティで `"optimizePerformance"` を True に設定します。カタログを読み取らない限り、`withSchema` を指定する必要もあります。`withSchema` は、「[XML スキーマを手動で指定する](aws-glue-programming-etl-format-xml-home.md#aws-glue-programming-etl-format-xml-withschema)」で説明されている入力を期待します ``` // Read from S3 data source glueContext.create_dynamic_frame.from_options( connection_type = "s3", connection_options = {"paths": ["s3://s3path"]}, format = "json", format_options={ "optimizePerformance": True, "withSchema": SchemaString }) // Read from catalog table glueContext.create_dynamic_frame.from_catalog( database = database, table_name = table, additional_options = { // The vectorized reader for JSON can read your schema from a catalog table property. "optimizePerformance": True, }) ``` AWS Glue ライブラリで *SchemaString* を構築する方法の詳細については、「[PySpark 拡張子型](aws-glue-api-crawler-pyspark-extensions-types.md)」を参照してください。 **ベクトル化された CSV リーダーでの制限事項** 以下の制限事項に留意してください。 + ネストされたオブジェクトまたは配列値を持つ JSON 要素はサポートされていません。指定されている場合、AWS Glue は標準のリーダーにフォールバックします。 + カタログから、または `withSchema` で、スキーマを指定する必要があります。 + `multiline` または `jsonPath` と互換性がありません。これらのオプションのいずれかを指定すると、AWS Glue は標準のリーダーにフォールバックするように指示されます。 + 入力スキーマと一致しない入力レコードを指定すると、リーダーは失敗します。 + [エラーレコード](https://docs.aws.amazon.com/glue/latest/dg/glue-etl-scala-apis-glue-dynamicframe-class.html#glue-etl-scala-apis-glue-dynamicframe-class-defs-errorsAsDynamicFrame)は作成されません。 + マルチバイト文字 (日本語や中国語の文字など) を含む JSON ファイルはサポートされていません。 # AWS Glue で ORC 形式を使用する AWS Glue はソースからデータを取得し、さまざまなデータ形式で保存および転送されたターゲットにデータを書き込みます。このドキュメントでは、データが ORC データ形式で保存または転送される場合に、AWS Glue でデータを使用する際に利用できる機能について説明します。 AWS Glue は、ORC 形式の使用をサポートしています。この形式は、パフォーマンス指向の列ベースのデータ形式です。標準局による形式の概要については、「[Apache Orc](https://orc.apache.org/docs/)」を参照してください。 AWS Glue を使用して、Amazon S3 およびストリーミングソースから ORC ファイルを読み取り、Amazon S3 に ORC ファイルを書き込むことができます。S3 から、ORC ファイルを含む `bzip` および `gzip` アーカイブを読み書きすることができます。このページで説明する設定ではなく、[S3 接続パラメータ](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) 上で圧縮動作を設定します。 次の表は、ORC 形式のオプションをサポートする一般的な AWS Glue の機能を示しています。 | 読み込み | 書き込み | ストリーミングの読み取り | 小さなファイルのグループ化 | ジョブのブックマーク | | --- | --- | --- | --- | --- | | サポート | サポート対象 | サポート | サポートされていません | サポート対象\$1 | \$1AWS Glue バージョン 1.0\$1 でサポート ## 例: S3 から ORC ファイルまたはフォルダを読み取る **前提条件:** 読み取る ORC ファイルまたはフォルダへの S3 パス (`s3path`) が必要です。 **設定:** 関数オプションで `format="orc"` を指定します。`connection_options` で、`paths` キーを使用して `s3path` を指定します。リーダーが S3 とやり取りする方法は、`connection_options` で設定できます。詳細については、AWS Glue: [Amazon S3 接続のオプションのリファレンス](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) の「ETL の接続タイプとオプション」を参照してください。 次の AWS Glue ETL スクリプトは、S3 から ORC ファイルまたはフォルダを読み取るプロセスを示しています。 ------ #### [ Python ] この例では [create\$1dynamic\$1frame.from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-create_dynamic_frame_from_options) メソッドを使用します。 ``` from pyspark.context import SparkContext from awsglue.context import GlueContext sc = SparkContext.getOrCreate() glueContext = GlueContext(sc) dynamicFrame = glueContext.create_dynamic_frame.from_options( connection_type="s3", connection_options={"paths": ["s3://s3path"]}, format="orc" ) ``` スクリプト (`pyspark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` dataFrame = spark.read\ .orc("s3://s3path") ``` ------ #### [ Scala ] この例では [getSourceWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSourceWithFormat) 操作を使用します。 ``` import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.GlueContext import org.apache.spark.sql.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) val dynamicFrame = glueContext.getSourceWithFormat( connectionType="s3", format="orc", options=JsonOptions("""{"paths": ["s3://s3path"]}""") ).getDynamicFrame() } } ``` スクリプト (`pyspark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` val dataFrame = spark.read .orc("s3://s3path") ``` ------ ## 例: ORC ファイルおよびフォルダを S3 に書き込む **前提条件:** 初期化された DataFrame (`dataFrame`) または DynamicFrame (`dynamicFrame`) が必要です。また、予想される S3 出力パスである `s3path` も必要になります。 **設定:** 関数オプションで `format="orc"` を指定します。接続オプションでは、`s3path` を指定するための `paths` キーを使用します。ライターが S3 と対話する方法を、`connection_options` でさらに詳しく変更することができます。詳細については、AWS Glue: [Amazon S3 接続のオプションのリファレンス](aws-glue-programming-etl-connect-s3-home.md#aws-glue-programming-etl-connect-s3) の「ETL の入力および出力のデータ形式オプション」を参照してください。次のコード例は、プロセスを示しています。 ------ #### [ Python ] この例では [write\$1dynamic\$1frame.from\$1options](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-write_dynamic_frame_from_options) メソッドを使用します。 ``` from pyspark.context import SparkContext from awsglue.context import GlueContext sc = SparkContext.getOrCreate() glueContext = GlueContext(sc) glueContext.write_dynamic_frame.from_options( frame=dynamicFrame, connection_type="s3", format="orc", connection_options={ "path": "s3://s3path" } ) ``` スクリプト (`pyspark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` df.write.orc("s3://s3path/") ``` ------ #### [ Scala ] この例では [getSinkWithFormat](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getSinkWithFormat) メソッドを使用します。 ``` import com.amazonaws.services.glue.util.JsonOptions import com.amazonaws.services.glue.{DynamicFrame, GlueContext} import org.apache.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) glueContext.getSinkWithFormat( connectionType="s3", options=JsonOptions("""{"path": "s3://s3path"}"""), format="orc" ).writeDynamicFrame(dynamicFrame) } } ``` スクリプト (`pyspark.sql.DataFrame`) では DataFrame を使用することもできます。 ``` df.write.orc("s3://s3path/") ``` ------ ## ORC 設定リファレンス `format="orc"` の `format_options` 値はありません。ただし、基になる SparkSQL コードで受け入れられるオプションは、`connection_options` マップパラメータを介して渡すことができます。 # AWS Glue ETL ジョブでのデータレイクフレームワークの使用 オープンソースのデータレイクフレームワークを使用するいことで、Amazon S3 上に構築されたデータレイクに保存するファイルの増分データ処理を簡素化できます。AWS Glue 3.0 以降で、以下のオープンソースデータレイクフレームワークがサポートされています。 + Apache Hudi + Linux Foundation Delta Lake + Apache Iceberg これらのフレームワークはネイティブにサポートされているので、Amazon S3 に保存するデータに対する読み書きが、一貫性のあるトランザクションにより実行できます。これらのフレームワークを AWS Glue ETLジョブで使用する場合にも、別のコネクタをインストールしたり、追加の構成手順を実行したりする必要はありません。 AWS Glue Data Catalog を介してデータセットを管理していれば、Spark DataFrames を使用してデータレイクテーブルを読み書きする場合に、AWS Glue メソッドを利用できます。また、Spark DataFrame API を使用して、Amazon S3 データを読み書きすることもできます。 このビデオでは、Apache Hudi、Apache Iceberg、Delta Lake の仕組みの基本について学ぶことができます。データレイクにデータを挿入、更新、削除する方法と、これらの各フレームワークの仕組みについて説明します。 [![AWS Videos](http://img.youtube.com/vi/https://www.youtube.com/embed/fryfx0Zg7KA/0.jpg)](http://www.youtube.com/watch?v=https://www.youtube.com/embed/fryfx0Zg7KA) **Topics** + [制限](aws-glue-programming-etl-datalake-native-frameworks-limitations.md) + [AWS Glue での Hudi フレームワークの使用](aws-glue-programming-etl-format-hudi.md) + [AWS Glue での Delta Lake フレームワークの使用](aws-glue-programming-etl-format-delta-lake.md) + [AWS Glue での Iceberg フレームワークの使用](aws-glue-programming-etl-format-iceberg.md) # 制限 AWS Glue でデータレイクフレームワークを使用する際には、以下の制限を考慮してください。 + DynamicFrame の次の AWS Glue `GlueContext` メソッドは、データレイクフレームワークテーブルの読み取りと書き込みをサポートしていません。代わりに、DataFrame または Spark DataFrame API については `GlueContext` メソッドを使用します。 + `create_dynamic_frame.from_catalog` + `write_dynamic_frame.from_catalog` + `getDynamicFrame` + `writeDynamicFrame` + DataFrame の次の `GlueContext` メソッドは、Lake Formation の許可のコントロールでサポートされています。 + `create_data_frame.from_catalog` + `write_data_frame.from_catalog` + `getDataFrame` + `writeDataFrame` + [小さなファイルのグループ化](grouping-input-files.md)は、サポートされません。 + [ジョブのブックマーク](monitor-continuations.md)はサポートされません。 + Apache Hudi 0.10.1 for AWS Glue 3.0 では、Hudi Merge on Read (MoR) テーブルはサポートされていません。 + Apache Iceberg 0.13.1 for AWS Glue 3.0 では、`ALTER TABLE … RENAME TO` を使用できません。 ## Lake Formation の許可によって管理されるデータレイク形式のテーブルの制限事項 データレイク形式は、Lake Formation の許可を介して AWS Glue ETL と統合されます。`create_dynamic_frame` を使用した DynamicFrame の作成はサポートされていません。詳細については、次の例を参照してください。 + [例: Lake Formation の許可のコントロールを使用した Iceberg テーブルの読み取りおよび書き込み](https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-format-iceberg.html#aws-glue-programming-etl-format-iceberg-read-write-lake-formation-tables) + [例: Lake Formation の許可のコントロールを使用した Hudi テーブルの読み取りおよび書き込み](https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-format-hudi.html#aws-glue-programming-etl-format-hudi-read-write-lake-formation-tables) + [例: Lake Formation の許可のコントロールを使用した Delta Lake テーブルの読み取りおよび書き込み](https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-format-delta-lake.html#aws-glue-programming-etl-format-delta-lake-read-write-lake-formation-tables) **注記** Apache Hudi、Apache Iceberg、Delta Lake の Lake Formation 許可による AWS Glue ETLとの統合は、AWS Glue バージョン 4.0 でのみサポートされています。 Apache Iceberg は、Lake Formation の許可を介して AWS Glue ETL と最適に統合されています。ほぼすべてのオペレーションをサポートしており、SQL サポートも含まれています。 Hudi は、管理オペレーションを除くほとんどの基本オペレーションをサポートします。これは、これらのオプションが通常、データフレームの書き込みを介して実行され、`additional_options` を介して指定されるためです。SparkSQL はサポートされていないため、オペレーションのために AWS Glue API を使用して DataFrame を作成する必要があります。 Delta Lake は、テーブルデータの読み取り、付加、および上書きのみをサポートします。Delta Lake では、更新などのさまざまなタスクを実行できるように独自のライブラリを使用する必要があります。 次の機能は、Lake Formation の許可によって管理される Iceberg テーブルでは使用できません。 + AWS Glue ETL を使用した圧縮 + AWS Glue ETL を介した Spark SQL サポート Lake Formation の許可によって管理される Hudi テーブルの制限は次のとおりです。 + 孤立したファイルの削除 Lake Formation の許可によって管理される Delta Lake テーブルの制限は次のとおりです。 + Delta Lake テーブルの挿入と読み取り以外のすべての機能。 # AWS Glue での Hudi フレームワークの使用 AWS Glue 3.0 以降では、データレイク向けに Apache Hudi フレームワークが利用できます。Hudi は、増分データ処理とデータパイプラインの開発を簡素化する、オープンソースのデータレイク用ストレージフレームワークです。このトピックでは、AWS Glue 内でデータを Hudi テーブルに転送または保存する際に利用可能な、各機能について説明します。Hudi の詳細については、公式の[Apache Hudi ドキュメント](https://hudi.apache.org/docs/overview/)を参照してください。 AWS Glue により、Amazon S3 内にある Hudi テーブルの読み取りおよび書き込み操作を実行できます。あるいは、AWS Glue データカタログを使用して、Hudi テーブルを操作することも可能です。挿入、更新、および、すべての [Apache Spark オペレーション](https://hudi.apache.org/docs/quick-start-guide/)を含む操作も、追加でサポートされています。 **注記** AWS Glue 5.0 の Apache Hudi 0.15.0 実装は [HUDI-7001](https://github.com/apache/hudi/pull/9936) を内部的に元に戻します。レコードキーが 1 つのフィールドで構成されている場合、Complex Key 生成に関連するリグレッションは表示されません。ただし、この動作は OSS Apache Hudi 0.15.0 とは異なります。 Apache Hudi 0.10.1 for AWS Glue 3.0 では、Hudi Merge on Read (MoR) テーブルはサポートされません。 次の表に、AWS Glue の各バージョンに含まれている、Hudi のバージョンを一覧で示します。 **** | AWS Glue のバージョン | サポートされる Hudi バージョン | | --- | --- | | 5.1 | 1.0.2 | | 5.0 | 0.15.0 | | 4.0 | 0.12.1 | | 3.0 | 0.10.1 | AWS Glue がサポートするデータレイクフレームワークの詳細については、「[AWS Glue ETL ジョブでのデータレイクフレームワークの使用](aws-glue-programming-etl-datalake-native-frameworks.md)」を参照してください。 ## Hudi の有効化 AWS Glue で Hudi を有効化するには、以下のタスクを実行します。 + `hudi` を `--datalake-formats` のジョブパラメータの値として指定します。詳細については、「[AWS Glue ジョブでジョブパラメータを使用する](aws-glue-programming-etl-glue-arguments.md)」を参照してください。 + AWS Glue ジョブ用に、`--conf` という名前でキーを作成し、それに次の値を設定します。または、スクリプトで `SparkConf` を使用して、次の構成を設定することもできます。これらの設定は、Apache Spark が Hudi テーブルを適切に処理するために役立ちます。 ``` spark.serializer=org.apache.spark.serializer.KryoSerializer ``` + Hudi に関する Lake Formation の許可のサポートは、AWS Glue 4.0 のためにデフォルトで有効になっています。Lake Formation に登録された Hudi テーブルの読み取り/書き込みに追加の設定は必要ありません。登録された Hudi テーブルを読み取るには、AWS Glue ジョブの IAM ロールに SELECT 許可が必要です。登録された Hudi テーブルに書き込むには、AWS Glue ジョブの IAM ロールに SUPER 許可が必要です。Lake Formation の許可の管理の詳細については、「[Data Catalog リソースに対する許可の付与と取り消し](https://docs.aws.amazon.com/lake-formation/latest/dg/granting-catalog-permissions.html)」を参照してください。 **別の Hudi バージョンの使用** AWS Glue でサポートされないバージョンの Hudi を使用するには、`--extra-jars` job パラメーターにより独自の Hudi JAR ファイルを指定します。`--datalake-formats` ジョブパラメータの値として、`hudi` は含めないでください。AWS Glue 5.0 以降を使用する場合、`--user-jars-first true` ジョブパラメータを設定する必要があります。 ## 例: Hudi テーブルを Amazon S3 に書き込み、そのテーブルを AWS Glue データカタログに登録する このスクリプト例では、Hudi テーブルを Amazon S3 に書き込み、そのテーブルを AWS Glue データカタログに登録する方法を示します。この例では、テーブルを登録するために、Hudi [Hive Sync ツール](https://hudi.apache.org/docs/syncing_metastore/)を使用します。 **注記** この例では、AWS Glue データカタログを Apache Spark Hive のメタストアとして使用するために、`--enable-glue-datacatalog` ジョブパラメータの設定が必要となります。詳細については[AWS Glue ジョブでジョブパラメータを使用する](aws-glue-programming-etl-glue-arguments.md)を参照してください。 ------ #### [ Python ] ``` # Example: Create a Hudi table from a DataFrame # and register the table to Glue Data Catalog additional_options={ "hoodie.table.name": "", "hoodie.database.name": "", "hoodie.datasource.write.storage.type": "COPY_ON_WRITE", "hoodie.datasource.write.operation": "upsert", "hoodie.datasource.write.recordkey.field": "", "hoodie.datasource.write.precombine.field": "", "hoodie.datasource.write.partitionpath.field": "", "hoodie.datasource.write.hive_style_partitioning": "true", "hoodie.datasource.hive_sync.enable": "true", "hoodie.datasource.hive_sync.database": "", "hoodie.datasource.hive_sync.table": "", "hoodie.datasource.hive_sync.partition_fields": "", "hoodie.datasource.hive_sync.partition_extractor_class": "org.apache.hudi.hive.MultiPartKeysValueExtractor", "hoodie.datasource.hive_sync.use_jdbc": "false", "hoodie.datasource.hive_sync.mode": "hms", "path": "s3://" } dataFrame.write.format("hudi") \ .options(**additional_options) \ .mode("overwrite") \ .save() ``` ------ #### [ Scala ] ``` // Example: Example: Create a Hudi table from a DataFrame // and register the table to Glue Data Catalog val additionalOptions = Map( "hoodie.table.name" -> "", "hoodie.database.name" -> "", "hoodie.datasource.write.storage.type" -> "COPY_ON_WRITE", "hoodie.datasource.write.operation" -> "upsert", "hoodie.datasource.write.recordkey.field" -> "", "hoodie.datasource.write.precombine.field" -> "", "hoodie.datasource.write.partitionpath.field" -> "", "hoodie.datasource.write.hive_style_partitioning" -> "true", "hoodie.datasource.hive_sync.enable" -> "true", "hoodie.datasource.hive_sync.database" -> "", "hoodie.datasource.hive_sync.table" -> "", "hoodie.datasource.hive_sync.partition_fields" -> "", "hoodie.datasource.hive_sync.partition_extractor_class" -> "org.apache.hudi.hive.MultiPartKeysValueExtractor", "hoodie.datasource.hive_sync.use_jdbc" -> "false", "hoodie.datasource.hive_sync.mode" -> "hms", "path" -> "s3://") dataFrame.write.format("hudi") .options(additionalOptions) .mode("append") .save() ``` ------ ## 例: AWS Glue データカタログを使用した Amazon S3 からの Hudi テーブルの読み取り この例では、「[例: Hudi テーブルを Amazon S3 に書き込み、そのテーブルを AWS Glue データカタログに登録する](#aws-glue-programming-etl-format-hudi-write)」で作成した Hudi テーブルを Amazon S3 から読み取ります。 **注記** この例では、AWS Glue データカタログを Apache Spark Hive のメタストアとして使用するために、`--enable-glue-datacatalog` ジョブパラメータの設定が必要となります。詳細については[AWS Glue ジョブでジョブパラメータを使用する](aws-glue-programming-etl-glue-arguments.md)を参照してください。 ------ #### [ Python ] この例では、`GlueContext.create\$1data\$1frame.from\$1catalog()` メソッドを使用します。 ``` # Example: Read a Hudi table from Glue Data Catalog from awsglue.context import GlueContext from pyspark.context import SparkContext sc = SparkContext() glueContext = GlueContext(sc) dataFrame = glueContext.create_data_frame.from_catalog( database = "", table_name = "" ) ``` ------ #### [ Scala ] この例では [getCatalogSource](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getCatalogSource) メソッドを使用します。 ``` // Example: Read a Hudi table from Glue Data Catalog import com.amazonaws.services.glue.GlueContext import org.apache.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) val dataFrame = glueContext.getCatalogSource( database = "", tableName = "" ).getDataFrame() } } ``` ------ ## 例: `DataFrame` を更新して、Amazon S3 内にある Hudi テーブルに挿入する この例では、「[例: Hudi テーブルを Amazon S3 に書き込み、そのテーブルを AWS Glue データカタログに登録する](#aws-glue-programming-etl-format-hudi-write)」で作成した Hudi テーブルに対し、AWS Glue データカタログを使用して DataFrame を挿入します。 **注記** この例では、AWS Glue データカタログを Apache Spark Hive のメタストアとして使用するために、`--enable-glue-datacatalog` ジョブパラメータの設定が必要となります。詳細については[AWS Glue ジョブでジョブパラメータを使用する](aws-glue-programming-etl-glue-arguments.md)を参照してください。 ------ #### [ Python ] この例では、`GlueContext.write\$1data\$1frame.from\$1catalog()` メソッドを使用します。 ``` # Example: Upsert a Hudi table from Glue Data Catalog from awsglue.context import GlueContext from pyspark.context import SparkContext sc = SparkContext() glueContext = GlueContext(sc) glueContext.write_data_frame.from_catalog( frame = dataFrame, database = "", table_name = "", additional_options={ "hoodie.table.name": "", "hoodie.database.name": "", "hoodie.datasource.write.storage.type": "COPY_ON_WRITE", "hoodie.datasource.write.operation": "upsert", "hoodie.datasource.write.recordkey.field": "", "hoodie.datasource.write.precombine.field": "", "hoodie.datasource.write.partitionpath.field": "", "hoodie.datasource.write.hive_style_partitioning": "true", "hoodie.datasource.hive_sync.enable": "true", "hoodie.datasource.hive_sync.database": "", "hoodie.datasource.hive_sync.table": "", "hoodie.datasource.hive_sync.partition_fields": "", "hoodie.datasource.hive_sync.partition_extractor_class": "org.apache.hudi.hive.MultiPartKeysValueExtractor", "hoodie.datasource.hive_sync.use_jdbc": "false", "hoodie.datasource.hive_sync.mode": "hms" } ) ``` ------ #### [ Scala ] この例では [getCatalogSink](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getCatalogSink) メソッドを使用します。 ``` // Example: Upsert a Hudi table from Glue Data Catalog import com.amazonaws.services.glue.GlueContext import com.amazonaws.services.glue.util.JsonOptions import org.apacke.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) glueContext.getCatalogSink("", "", additionalOptions = JsonOptions(Map( "hoodie.table.name" -> "", "hoodie.database.name" -> "", "hoodie.datasource.write.storage.type" -> "COPY_ON_WRITE", "hoodie.datasource.write.operation" -> "upsert", "hoodie.datasource.write.recordkey.field" -> "", "hoodie.datasource.write.precombine.field" -> "", "hoodie.datasource.write.partitionpath.field" -> "", "hoodie.datasource.write.hive_style_partitioning" -> "true", "hoodie.datasource.hive_sync.enable" -> "true", "hoodie.datasource.hive_sync.database" -> "", "hoodie.datasource.hive_sync.table" -> "", "hoodie.datasource.hive_sync.partition_fields" -> "", "hoodie.datasource.hive_sync.partition_extractor_class" -> "org.apache.hudi.hive.MultiPartKeysValueExtractor", "hoodie.datasource.hive_sync.use_jdbc" -> "false", "hoodie.datasource.hive_sync.mode" -> "hms" ))) .writeDataFrame(dataFrame, glueContext) } } ``` ------ ## 例: Spark を使用した Amazon S3 からの Hudi テーブルの読み取り この例では、Spark DataFrame API を使用して Amazon S3 から Hudi テーブルを読み取ります。 ------ #### [ Python ] ``` # Example: Read a Hudi table from S3 using a Spark DataFrame dataFrame = spark.read.format("hudi").load("s3://") ``` ------ #### [ Scala ] ``` // Example: Read a Hudi table from S3 using a Spark DataFrame val dataFrame = spark.read.format("hudi").load("s3://") ``` ------ ## 例: スパークを使用した Amazon S3 への Hudi テーブルの書き込み この例では、Spark を使用して Amazon S3 に Hudi テーブルを書き込みます。 ------ #### [ Python ] ``` # Example: Write a Hudi table to S3 using a Spark DataFrame dataFrame.write.format("hudi") \ .options(**additional_options) \ .mode("overwrite") \ .save("s3://) ``` ------ #### [ Scala ] ``` // Example: Write a Hudi table to S3 using a Spark DataFrame dataFrame.write.format("hudi") .options(additionalOptions) .mode("overwrite") .save("s3://") ``` ------ ## 例: Lake Formation の許可のコントロールを使用した Hudi テーブルの読み取りおよび書き込み この例では、Lake Formation の許可のコントロールを使用して Hudi テーブルの読み取りと書き込みを行います。 1. Hudi テーブルを作成して Lake Formation に登録します。 1. Lake Formation の許可のコントロールを有効にするには、まずテーブルの Amazon S3 パスを Lake Formation に登録する必要があります。詳細については、「[Registering an Amazon S3 location](https://docs.aws.amazon.com/lake-formation/latest/dg/register-location.html)」を参照してください。Lake Formation コンソールから、または AWS CLI を使用して登録できます。 ``` aws lakeformation register-resource --resource-arn arn:aws:s3:::/ --use-service-linked-role --region ``` Amazon S3 の場所が登録されると、その場所 (またはその子である場所) をポイントするすべての AWS Glue テーブルが、`GetTable` 呼び出しで `IsRegisteredWithLakeFormation` パラメータの値を true として返します。 1. Spark DataFrame API を介して登録された Amazon S3 パスをポイントする Hudi テーブルを作成します。 ``` hudi_options = { 'hoodie.table.name': table_name, 'hoodie.database.name': database_name, 'hoodie.datasource.write.storage.type': 'COPY_ON_WRITE', 'hoodie.datasource.write.recordkey.field': 'product_id', 'hoodie.datasource.write.table.name': table_name, 'hoodie.datasource.write.operation': 'upsert', 'hoodie.datasource.write.precombine.field': 'updated_at', 'hoodie.datasource.write.hive_style_partitioning': 'true', 'hoodie.upsert.shuffle.parallelism': 2, 'hoodie.insert.shuffle.parallelism': 2, 'path': , 'hoodie.datasource.hive_sync.enable': 'true', 'hoodie.datasource.hive_sync.database': database_name, 'hoodie.datasource.hive_sync.table': table_name, 'hoodie.datasource.hive_sync.use_jdbc': 'false', 'hoodie.datasource.hive_sync.mode': 'hms' } df_products.write.format("hudi") \ .options(**hudi_options) \ .mode("overwrite") \ .save() ``` 1. AWS Glue ジョブ IAM ロールに Lake Formation の許可を付与します。Lake Formation コンソールから、または AWS CLI を使用して許可を付与できます。詳細については、「[Lake Formation コンソールと名前付きリソース方式を使用したテーブル許可の付与](https://docs.aws.amazon.com/lake-formation/latest/dg/granting-table-permissions.html)」を参照してください。 1. Lake Formation に登録されている Hudi テーブルを読み取ります。このコードは、未登録の Hudi テーブルを読み取る場合と同じです。読み取りを成功させるには、AWS Glue ジョブの IAM ロールに SELECT 許可が必要であることに留意してください。 ``` val dataFrame = glueContext.getCatalogSource( database = "", tableName = "" ).getDataFrame() ``` 1. Lake Formation に登録されている Hudi テーブルに書き込みます。このコードは、未登録の Hudi テーブルに書き込む場合と同じです。書き込みを成功させるには、AWS Glue ジョブの IAM ロールに SUPER 許可が必要であることに留意してください。 ``` glueContext.getCatalogSink("", "", additionalOptions = JsonOptions(Map( "hoodie.table.name" -> "", "hoodie.database.name" -> "", "hoodie.datasource.write.storage.type" -> "COPY_ON_WRITE", "hoodie.datasource.write.operation" -> "", "hoodie.datasource.write.recordkey.field" -> "", "hoodie.datasource.write.precombine.field" -> "", "hoodie.datasource.write.partitionpath.field" -> "", "hoodie.datasource.write.hive_style_partitioning" -> "true", "hoodie.datasource.hive_sync.enable" -> "true", "hoodie.datasource.hive_sync.database" -> "", "hoodie.datasource.hive_sync.table" -> "", "hoodie.datasource.hive_sync.partition_fields" -> "", "hoodie.datasource.hive_sync.partition_extractor_class" -> "org.apache.hudi.hive.MultiPartKeysValueExtractor", "hoodie.datasource.hive_sync.use_jdbc" -> "false", "hoodie.datasource.hive_sync.mode" -> "hms" ))) .writeDataFrame(dataFrame, glueContext) ``` # AWS Glue での Delta Lake フレームワークの使用 AWS Glue 3.0 以降では、Linux Foundation Delta Lake フレームワークを利用できます。Delta Lake は、ACID トランザクションの実行、メタデータ処理のスケーリング、さらにストリーミングとバッチデータ処理の統合を支援する、オープンソースのデータレイクストレージフレームワークです。このトピックでは、Delta Lake テーブルに転送または保存するデータに対して、AWS Glue 内で利用可能な機能について説明します。Delta Lake の詳細については、公式の [Delta Lake のドキュメント](https://docs.delta.io/latest/delta-intro.html)を参照してください。 Amazon S3 内にある Delta Lake テーブルに対する読み取りおよび書き込みの操作は、AWS Glue を使用して実行できます。あるいは、AWS Glue データカタログにより Delta Lake テーブルを操作することもできます。挿入、更新、および[テーブルに対する一括での読み取りと書き込み](https://docs.delta.io/0.7.0/api/python/index.html)など、その他の操作もサポートされています。Delta Lake テーブルを使用する場合、Delta Lake Python ライブラリに備わったメソッド (例えば、`DeltaTable.forPath`) も使用可能です。Delta Lake Python ライブラリの詳細については、「Delta Lake の Python ドキュメント」を参照してください。 次の表に、AWS Glue の各バージョンに含まれている Delta Lake のバージョンを一覧で示します。 **** | AWS Glue のバージョン | サポートされている Delta Lake バージョン | | --- | --- | | 5.1 | 3.3.2 | | 5.0 | 3.3.0 | | 4.0 | 2.1.0 | | 3.0 | 1.0.0 | AWS Glue がサポートするデータレイクフレームワークの詳細については、「[AWS Glue ETL ジョブでのデータレイクフレームワークの使用](aws-glue-programming-etl-datalake-native-frameworks.md)」を参照してください。 ## AWS Glue のための Delta Lake の有効化 AWS Glue で Delta Lake を有効化するには、以下のタスクを実行します。 + `delta` を `--datalake-formats` のジョブパラメータの値として指定します。詳細については、「[AWS Glue ジョブでジョブパラメータを使用する](aws-glue-programming-etl-glue-arguments.md)」を参照してください。 + AWS Glue ジョブ用に、`--conf` という名前でキーを作成し、それに次の値を設定します。または、スクリプトで `SparkConf` を使用して、次の構成を設定することもできます。これらの設定は、Apache Spark が Delta Lake テーブルを適切に処理するために役立ちます。 ``` spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog --conf spark.delta.logStore.class=org.apache.spark.sql.delta.storage.S3SingleDriverLogStore ``` + Delta テーブルに関する Lake Formation の許可のサポートは、AWS Glue 4.0 のためにデフォルトで有効になっています。Lake Formation に登録された Delta テーブルの読み取り/書き込みに追加の設定は必要ありません。登録された Delta テーブルを読み取るには、AWS Glue ジョブの IAM ロールに SELECT 許可が必要です。登録された Delta テーブルに書き込むには、AWS Glue ジョブの IAM ロールに SUPER 許可が必要です。Lake Formation の許可の管理の詳細については、「[Data Catalog リソースに対する許可の付与と取り消し](https://docs.aws.amazon.com/lake-formation/latest/dg/granting-catalog-permissions.html)」を参照してください。 **別のバージョンの Delta Lake を使用する** AWS Glue でサポートされないバージョンの Delta Lake を使用するには、`--extra-jars` ジョブパラメータにより、独自の Delta Lake JAR ファイルを指定します。`--datalake-formats` ジョブパラメータの値として、`delta` は含めないでください。AWS Glue 5.0 以降を使用する場合、`--user-jars-first true` ジョブパラメータを設定する必要があります。この場合、Delta Lake Python ライブラリを使用するには、`--extra-py-files` ジョブパラメータにより、ライブラリの JAR ファイルを指定する必要があります。Python ライブラリは、Delta Lake JAR ファイルにパッケージ化されています。 ## 例: Delta Lake テーブルを Amazon S3 に書き込んで、そのテーブルを AWS Glue データカタログに登録する 次の AWS Glue ETL スクリプトは、Delta Lake テーブルを Amazon S3 に書き込み、そのテーブルを AWS Glue データカタログに登録する方法を示しています。 ------ #### [ Python ] ``` # Example: Create a Delta Lake table from a DataFrame # and register the table to Glue Data Catalog additional_options = { "path": "s3://" } dataFrame.write \ .format("delta") \ .options(**additional_options) \ .mode("append") \ .partitionBy("") \ .saveAsTable(".") ``` ------ #### [ Scala ] ``` // Example: Example: Create a Delta Lake table from a DataFrame // and register the table to Glue Data Catalog val additional_options = Map( "path" -> "s3://" ) dataFrame.write.format("delta") .options(additional_options) .mode("append") .partitionBy("") .saveAsTable(".") ``` ------ ## 例: AWS Glue データカタログを使用して Amazon S3 から Delta Lake テーブルを読み取る 次の AWS Glue ETL スクリプトでは、「[例: Delta Lake テーブルを Amazon S3 に書き込んで、そのテーブルを AWS Glue データカタログに登録する](#aws-glue-programming-etl-format-delta-lake-write)」で作成した Delta Lake テーブルの読み取りを実行します。 ------ #### [ Python ] この例では [create\$1data\$1frame.from\$1catalog](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-create-dataframe-from-catalog) メソッドを使用します。 ``` # Example: Read a Delta Lake table from Glue Data Catalog from awsglue.context import GlueContext from pyspark.context import SparkContext sc = SparkContext() glueContext = GlueContext(sc) df = glueContext.create_data_frame.from_catalog( database="", table_name="", additional_options=additional_options ) ``` ------ #### [ Scala ] この例では [getCatalogSource](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getCatalogSource) メソッドを使用します。 ``` // Example: Read a Delta Lake table from Glue Data Catalog import com.amazonaws.services.glue.GlueContext import org.apacke.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) val df = glueContext.getCatalogSource("", "", additionalOptions = additionalOptions) .getDataFrame() } } ``` ------ ## 例: AWS Glue データカタログを使用して Amazon S3 内にある Delta Lake テーブルに `DataFrame` を挿入する この例では、「[例: Delta Lake テーブルを Amazon S3 に書き込んで、そのテーブルを AWS Glue データカタログに登録する](#aws-glue-programming-etl-format-delta-lake-write)」で作成した Delta Lake テーブルにデータを挿入します。 **注記** この例では、AWS Glue データカタログを Apache Spark Hive のメタストアとして使用するために、`--enable-glue-datacatalog` ジョブパラメータの設定が必要となります。詳細については[AWS Glue ジョブでジョブパラメータを使用する](aws-glue-programming-etl-glue-arguments.md)を参照してください。 ------ #### [ Python ] この例では、[write\$1data\$1frame.from\$1catalog](aws-glue-api-crawler-pyspark-extensions-glue-context.md#aws-glue-api-crawler-pyspark-extensions-glue-context-write_data_frame_from_catalog) メソッドを使用します。 ``` # Example: Insert into a Delta Lake table in S3 using Glue Data Catalog from awsglue.context import GlueContext from pyspark.context import SparkContext sc = SparkContext() glueContext = GlueContext(sc) glueContext.write_data_frame.from_catalog( frame=dataFrame, database="", table_name="", additional_options=additional_options ) ``` ------ #### [ Scala ] この例では [getCatalogSink](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getCatalogSink) メソッドを使用します。 ``` // Example: Insert into a Delta Lake table in S3 using Glue Data Catalog import com.amazonaws.services.glue.GlueContext import org.apacke.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) glueContext.getCatalogSink("", "", additionalOptions = additionalOptions) .writeDataFrame(dataFrame, glueContext) } } ``` ------ ## 例: Spark API を使用して Amazon S3 から Delta Lake テーブルを読み取る この例では、Spark API を使用して Amazon S3 から Delta Lake を読み取ります。 ------ #### [ Python ] ``` # Example: Read a Delta Lake table from S3 using a Spark DataFrame dataFrame = spark.read.format("delta").load("s3://") ``` ------ #### [ Scala ] ``` // Example: Read a Delta Lake table from S3 using a Spark DataFrame val dataFrame = spark.read.format("delta").load("s3://") ``` ------ ## 例: Spark を使用した Amazon S3 への Delta Lake テーブルの書き込み この例では、Spark を使用して Amazon S3 に Delta Lake テーブルを書き込みます。 ------ #### [ Python ] ``` # Example: Write a Delta Lake table to S3 using a Spark DataFrame dataFrame.write.format("delta") \ .options(**additional_options) \ .mode("overwrite") \ .partitionBy("") .save("s3://") ``` ------ #### [ Scala ] ``` // Example: Write a Delta Lake table to S3 using a Spark DataFrame dataFrame.write.format("delta") .options(additionalOptions) .mode("overwrite") .partitionBy("") .save("s3://") ``` ------ ## 例: Lake Formation の許可のコントロールを使用した Delta Lake テーブルの読み取りおよび書き込み この例では、Lake Formation の許可のコントロールを使用して Delta Lake テーブルの読み取りと書き込みを行います。 1. Delta テーブルを作成して Lake Formation に登録する 1. Lake Formation の許可のコントロールを有効にするには、まずテーブルの Amazon S3 パスを Lake Formation に登録する必要があります。詳細については、「[Registering an Amazon S3 location](https://docs.aws.amazon.com/lake-formation/latest/dg/register-location.html)」を参照してください。Lake Formation コンソールから、または AWS CLI を使用して登録できます。 ``` aws lakeformation register-resource --resource-arn arn:aws:s3:::/ --use-service-linked-role --region ``` Amazon S3 の場所が登録されると、その場所 (またはその子である場所) をポイントするすべての AWS Glue テーブルが、`GetTable` 呼び出しで `IsRegisteredWithLakeFormation` パラメータの値を true として返します。 1. Spark を介して登録された Amazon S3 パスをポイントする Delta テーブルを作成します。 **注記** 次に Python の例を示します。 ``` dataFrame.write \ .format("delta") \ .mode("overwrite") \ .partitionBy("") \ .save("s3://") ``` データが Amazon S3 に書き込まれた後、AWS Glue クローラーを使用して新しい Delta カタログテーブルを作成します。詳細については、「[AWS Glue クローラーによるネイティブデータレイクテーブルサポートの紹介](https://aws.amazon.com/blogs/big-data/introducing-native-delta-lake-table-support-with-aws-glue-crawlers/)」を参照してください。 AWS Glue `CreateTable` API を通じてテーブルを手動で作成することもできます。 1. AWS Glue ジョブ IAM ロールに Lake Formation の許可を付与します。Lake Formation コンソールから、または AWS CLI を使用して許可を付与できます。詳細については、「[Lake Formation コンソールと名前付きリソース方式を使用したテーブル許可の付与](https://docs.aws.amazon.com/lake-formation/latest/dg/granting-table-permissions.html)」を参照してください。 1. Lake Formation に登録されている Delta テーブルを読み取ります。このコードは、未登録の Delta テーブルを読み取る場合と同じです。読み取りを成功させるには、AWS Glue ジョブの IAM ロールに SELECT 許可が必要であることに留意してください。 ``` # Example: Read a Delta Lake table from Glue Data Catalog df = glueContext.create_data_frame.from_catalog( database="", table_name="", additional_options=additional_options ) ``` 1. Lake Formation に登録されている Delta テーブルに書き込みます。このコードは、未登録の Delta テーブルに書き込む場合と同じです。書き込みを成功させるには、AWS Glue ジョブの IAM ロールに SUPER 許可が必要であることに留意してください。 デフォルトでは、AWS Glue は saveMode として `Append` を使用します。これを変更するには、`additional_options` で saveMode オプションを設定します。Delta テーブルでの saveMode サポートの詳細については、「[テーブルに書き込む](https://docs.delta.io/latest/delta-batch.html#write-to-a-table)」を参照してください。 ``` glueContext.write_data_frame.from_catalog( frame=dataFrame, database="", table_name="", additional_options=additional_options ) ``` # AWS Glue での Iceberg フレームワークの使用 AWS Glue 3.0 以降では、データレイク向けに Apache Iceberg フレームワークがサポートされています。Iceberg により、SQL テーブルと同様な機能を持つ、高性能なテーブルフォーマットが提供されます。このトピックでは、AWS Glue 内でデータを Iceberg テーブルに転送または保存する際に利用可能な、機能について説明します。Iceberg の詳細については、公式の[Apache Iceberg ドキュメント](https://iceberg.apache.org/docs/latest/)を参照してください。 Amazon S3 内の Iceberg テーブルに対する読み取りと書き込みの操作は、AWS Glue を使用して実行することができます。あるいは、AWS Glue データカタログを使用して Iceberg テーブルを操作することも可能です。挿入に加え、「[Spark クエリ](https://iceberg.apache.org/docs/latest/spark-queries/)」 や 「[Spark 書き込み](https://iceberg.apache.org/docs/latest/spark-writes/)」 のすべての操作も、追加でサポートされています。更新は Iceberg テーブルではサポートされていません。 **注記** Apache Iceberg 0.13.1 for AWS Glue 3.0 では、`ALTER TABLE … RENAME TO` はサポートされません。 次の表に、AWS Glue の各バージョンに含まれる Iceberg のバージョンを一覧で示します。 **** | AWS Glue のバージョン | サポートされる Iceberg バージョン | | --- | --- | | 5.1 | 1.10.0 | | 5.0 | 1.7.1 | | 4.0 | 1.0.0 | | 3.0 | 0.13.1 | AWS Glue がサポートするデータレイクフレームワークの詳細については、「[AWS Glue ETL ジョブでのデータレイクフレームワークの使用](aws-glue-programming-etl-datalake-native-frameworks.md)」を参照してください。 ## Iceberg フレームワークの有効化 AWS Glue で Iceberg を有効化するには、以下のタスクを実行します。 + `iceberg` を `--datalake-formats` のジョブパラメータの値として指定します。詳細については、「[AWS Glue ジョブでジョブパラメータを使用する](aws-glue-programming-etl-glue-arguments.md)」を参照してください。 + AWS Glue ジョブ用に、`--conf` という名前でキーを作成し、それに次の値を設定します。または、スクリプトで `SparkConf` を使用して、次の構成を設定することもできます。これらの設定は、Apache Spark が Iceberg テーブルを適切に処理する際に役立ちます。 ``` spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions --conf spark.sql.catalog.glue_catalog=org.apache.iceberg.spark.SparkCatalog --conf spark.sql.catalog.glue_catalog.warehouse=s3:/// --conf spark.sql.catalog.glue_catalog.catalog-impl=org.apache.iceberg.aws.glue.GlueCatalog --conf spark.sql.catalog.glue_catalog.io-impl=org.apache.iceberg.aws.s3.S3FileIO ``` Lake Formation に登録されている Iceberg テーブルを読み書きする場合は、AWS Glue 5.0 以降の [AWS Glue と AWS Lake Formation を併用したきめ細かなアクセスコントロール](security-lf-enable.md) のガイダンスに従ってください。AWS Glue 4.0 で、次の設定を追加して Lake Formation のサポートを有効にします。 ``` --conf spark.sql.catalog.glue_catalog.glue.lakeformation-enabled=true --conf spark.sql.catalog.glue_catalog.glue.id= ``` AWS Glue 3.0 で Iceberg 0.13.1 を使用する場合には、Amazon DynamoDB ロックマネージャーを使用して確実なアトミックトランザクションを行うために、次の追加事項を設定する必要があります。 AWSGlue 4.0 以降では、デフォルトで楽観的ロックが使用されます。詳細については、公式の Apache Iceberg ドキュメントで「[Iceberg と AWS の統合](https://iceberg.apache.org/docs/latest/aws/#dynamodb-lock-manager)」を参照してください。 ``` --conf spark.sql.catalog.glue_catalog.lock-impl=org.apache.iceberg.aws.glue.DynamoLockManager --conf spark.sql.catalog.glue_catalog.lock.table= ``` **別バージョンの Iceberg の使用** AWS Glue でサポートされないバージョンの Iceberg を使用するには、`--extra-jars` ジョブパラメータを使用して独自の Iceberg JAR ファイルを指定します。`--datalake-formats` パラメータの値として、`iceberg` を含めないようにしてください。AWS Glue 5.0 以降を使用する場合、`--user-jars-first true` ジョブパラメータを設定する必要があります。 **Iceberg テーブルの暗号化を有効にする** **注記** Iceberg テーブルにはサーバー側の暗号化を有効にする独自のメカニズムがあります。AWS Glue のセキュリティ設定に加えて、これらの設定を有効にする必要があります。 Iceberg テーブルでサーバー側の暗号化を有効にするには、「[Iceberg ドキュメント](https://iceberg.apache.org/docs/latest/aws/#s3-server-side-encryption)」のガイダンスを確認してください。 **Iceberg クロスリージョンに Spark 設定を追加する** AWS Glue Data Catalog および AWS Lake Formation を使用して Iceberg クロスリージョンテーブルアクセスの Spark 設定を追加するには、以下の手順に従います。 1. [マルチリージョンアクセスポイント](https://docs.aws.amazon.com/AmazonS3/latest/userguide/multi-region-access-point-create-examples.html)を作成します。 1. 以下の Spark パラメータを設定します。 ``` ----- --conf spark.sql.catalog.my_catalog.s3.use-arn-region-enabled=true \ --conf spark.sql.catalog.{CATALOG}.s3.access-points.bucket1", "arn:aws:s3:::accesspoint/.mrap \ --conf spark.sql.catalog.{CATALOG}.s3.access-points.bucket2", "arn:aws:s3:::accesspoint/.mrap ----- ``` ## 例: Amazon S3 に Iceberg テーブルを書き込み、そのテーブルを AWS Glue データカタログに登録する このスクリプト例は、Amazon S3 に Iceberg テーブルを書き込む方法を示しています。この例では、[Iceberg と AWS の統合](https://iceberg.apache.org/docs/latest/aws/)を使用して、テーブルを AWS Glue データカタログに登録しています。 ------ #### [ Python ] ``` # Example: Create an Iceberg table from a DataFrame # and register the table to Glue Data Catalog dataFrame.createOrReplaceTempView("tmp_") query = f""" CREATE TABLE glue_catalog.. USING iceberg TBLPROPERTIES ("format-version"="2") AS SELECT * FROM tmp_ """ spark.sql(query) ``` ------ #### [ Scala ] ``` // Example: Example: Create an Iceberg table from a DataFrame // and register the table to Glue Data Catalog dataFrame.createOrReplaceTempView("tmp_") val query = """CREATE TABLE glue_catalog.. USING iceberg TBLPROPERTIES ("format-version"="2") AS SELECT * FROM tmp_ """ spark.sql(query) ``` ------ または、Spark メソッドを使用して Amazon S3 とデータカタログに Iceberg テーブルを書き込むこともできます。 前提条件: Iceberg ライブラリが使用するカタログを用意する必要があります。AWS Glue データカタログを使う場合、AWS Glue を使えば簡単です。AWS Glue データカタログは、`glue_catalog` として Spark ライブラリで使用できるように事前に設定されています。データカタログテーブルは、*atabaseName* と*tableName* で識別されます。AWS Glue データカタログの詳細については、「[AWS Glue でのデータ検出とカタログ化](catalog-and-crawler.md)」を参照してください AWS Glue データカタログを使用していない場合は、Spark API を使用してカタログをプロビジョニングする必要があります。詳細については、Spark ドキュメントの「[Spark の設定](https://iceberg.apache.org/docs/latest/spark-configuration/)」を参照してください。 この例では、Spark を使用して Amazon S3 と データカタログに Iceberg テーブルを書き込みます。 ------ #### [ Python ] ``` # Example: Write an Iceberg table to S3 on the Glue Data Catalog # Create (equivalent to CREATE TABLE AS SELECT) dataFrame.writeTo("glue_catalog.databaseName.tableName") \ .tableProperty("format-version", "2") \ .create() # Append (equivalent to INSERT INTO) dataFrame.writeTo("glue_catalog.databaseName.tableName") \ .tableProperty("format-version", "2") \ .append() ``` ------ #### [ Scala ] ``` // Example: Write an Iceberg table to S3 on the Glue Data Catalog // Create (equivalent to CREATE TABLE AS SELECT) dataFrame.writeTo("glue_catalog.databaseName.tableName") .tableProperty("format-version", "2") .create() // Append (equivalent to INSERT INTO) dataFrame.writeTo("glue_catalog.databaseName.tableName") .tableProperty("format-version", "2") .append() ``` ------ ## 例: AWS Glue データカタログを使用した Amazon S3 からの Iceberg テーブルの読み込み この例では、「[例: Amazon S3 に Iceberg テーブルを書き込み、そのテーブルを AWS Glue データカタログに登録する](#aws-glue-programming-etl-format-iceberg-write)」で作成した Iceberg テーブルを読み取っています。 ------ #### [ Python ] この例では、`GlueContext.create\$1data\$1frame.from\$1catalog()` メソッドを使用します。 ``` # Example: Read an Iceberg table from Glue Data Catalog from awsglue.context import GlueContext from pyspark.context import SparkContext sc = SparkContext() glueContext = GlueContext(sc) df = glueContext.create_data_frame.from_catalog( database="", table_name="", additional_options=additional_options ) ``` ------ #### [ Scala ] この例では [getCatalogSource](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getCatalogSource) メソッドを使用します。 ``` // Example: Read an Iceberg table from Glue Data Catalog import com.amazonaws.services.glue.GlueContext import org.apacke.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) val df = glueContext.getCatalogSource("", "", additionalOptions = additionalOptions) .getDataFrame() } } ``` ------ ## 例: AWS Glue データカタログを使用した Amazon S3 にある Iceberg テーブルへの `DataFrame` の挿入 この例では、「[例: Amazon S3 に Iceberg テーブルを書き込み、そのテーブルを AWS Glue データカタログに登録する](#aws-glue-programming-etl-format-iceberg-write)」で作成した Iceberg テーブルにデータを挿入します。 **注記** この例では、AWS Glue データカタログを Apache Spark Hive のメタストアとして使用するために、`--enable-glue-datacatalog` ジョブパラメータの設定が必要となります。詳細については[AWS Glue ジョブでジョブパラメータを使用する](aws-glue-programming-etl-glue-arguments.md)を参照してください。 ------ #### [ Python ] この例では、`GlueContext.write\$1data\$1frame.from\$1catalog()` メソッドを使用します。 ``` # Example: Insert into an Iceberg table from Glue Data Catalog from awsglue.context import GlueContext from pyspark.context import SparkContext sc = SparkContext() glueContext = GlueContext(sc) glueContext.write_data_frame.from_catalog( frame=dataFrame, database="", table_name="", additional_options=additional_options ) ``` ------ #### [ Scala ] この例では [getCatalogSink](glue-etl-scala-apis-glue-gluecontext.md#glue-etl-scala-apis-glue-gluecontext-defs-getCatalogSink) メソッドを使用します。 ``` // Example: Insert into an Iceberg table from Glue Data Catalog import com.amazonaws.services.glue.GlueContext import org.apacke.spark.SparkContext object GlueApp { def main(sysArgs: Array[String]): Unit = { val spark: SparkContext = new SparkContext() val glueContext: GlueContext = new GlueContext(spark) glueContext.getCatalogSink("", "", additionalOptions = additionalOptions) .writeDataFrame(dataFrame, glueContext) } } ``` ------ ## 例: Spark を使用した Iceberg テーブルの Amazon S3 からの読み取り 前提条件: Iceberg ライブラリが使用するカタログを用意する必要があります。AWS Glue データカタログを使う場合、AWS Glue を使えば簡単です。AWS Glue データカタログは、`glue_catalog` として Spark ライブラリで使用できるように事前に設定されています。データカタログテーブルは、*atabaseName* と*tableName* で識別されます。AWS Glue データカタログの詳細については、「[AWS Glue でのデータ検出とカタログ化](catalog-and-crawler.md)」を参照してください AWS Glue データカタログを使用していない場合は、Spark API を使用してカタログをプロビジョニングする必要があります。詳細については、Spark ドキュメントの「[Spark の設定](https://iceberg.apache.org/docs/latest/spark-configuration/)」を参照してください。 この例では、Spark を使用してデータカタログから Amazon S3 の Iceberg テーブルを読み取ります。 ------ #### [ Python ] ``` # Example: Read an Iceberg table on S3 as a DataFrame from the Glue Data Catalog dataFrame = spark.read.format("iceberg").load("glue_catalog.databaseName.tableName") ``` ------ #### [ Scala ] ``` // Example: Read an Iceberg table on S3 as a DataFrame from the Glue Data Catalog val dataFrame = spark.read.format("iceberg").load("glue_catalog.databaseName.tableName") ``` ------ ## 例: Lake Formation の許可のコントロールを使用した Iceberg テーブルの読み取りおよび書き込み この例では、Lake Formation の許可のコントロールを使用して Iceberg テーブルの読み取りと書き込みを行います。 **注記** この例は、AWS Glue 4.0 でのみ動作します。AWS Glue 5.0 以降では、「[AWS Glue と AWS Lake Formation を併用したきめ細かなアクセスコントロール](security-lf-enable.md)」のガイダンスに従ってください。 1. Iceberg テーブルを作成して Lake Formation に登録します: 1. Lake Formation の許可のコントロールを有効にするには、まずテーブルの Amazon S3 パスを Lake Formation に登録する必要があります。詳細については、「[Registering an Amazon S3 location](https://docs.aws.amazon.com/lake-formation/latest/dg/register-location.html)」を参照してください。Lake Formation コンソールから、または AWS CLI を使用して登録できます。 ``` aws lakeformation register-resource --resource-arn arn:aws:s3:::/ --use-service-linked-role --region ``` Amazon S3 の場所が登録されると、その場所 (またはその子である場所) をポイントするすべての AWS Glue テーブルが、`GetTable` 呼び出しで `IsRegisteredWithLakeFormation` パラメータの値を true として返します。 1. Spark SQL を介して登録されたパスをポイントする Iceberg テーブルを作成します。 **注記** 次に Python の例を示します。 ``` dataFrame.createOrReplaceTempView("tmp_") query = f""" CREATE TABLE glue_catalog.. USING iceberg AS SELECT * FROM tmp_ """ spark.sql(query) ``` AWS Glue `CreateTable` API を通じてテーブルを手動で作成することもできます。詳細については、「[Apache Iceberg テーブルの作成](https://docs.aws.amazon.com/lake-formation/latest/dg/creating-iceberg-tables.html)」を参照してください。 **注記** `UpdateTable` API は現在、 オペレーションへの入力として Iceberg テーブル形式をサポートしていません。 1. ジョブ IAM ロールに Lake Formation の許可を付与します。Lake Formation コンソールから、または AWS CLI を使用して許可を付与できます。詳細については、次を参照してください: https://docs.aws.amazon.com/lake-formation/latest/dg/granting-table-permissions.html 1. Lake Formation に登録されている Iceberg テーブルを読み取ります。このコードは、未登録の Iceberg テーブルを読み取る場合と同じです。読み取りを成功させるには、AWS Glue ジョブの IAM ロールに SELECT 許可が必要であることに留意してください。 ``` # Example: Read an Iceberg table from the AWS Glue Data Catalog from awsglue.context import GlueContextfrom pyspark.context import SparkContext sc = SparkContext() glueContext = GlueContext(sc) df = glueContext.create_data_frame.from_catalog( database="", table_name="", additional_options=additional_options ) ``` 1. Lake Formation に登録されている Iceberg テーブルに書き込みます。このコードは、未登録の Iceberg テーブルに書き込む場合と同じです。書き込みを成功させるには、AWS Glue ジョブの IAM ロールに SUPER 許可が必要であることに留意してください。 ``` glueContext.write_data_frame.from_catalog( frame=dataFrame, database="", table_name="", additional_options=additional_options ) ``` ## 共有設定リファレンス 形式タイプには次の `format_options` 値を使用できます。 + `attachFilename` — 列名として使用する適切な形式の文字列。このオプションを指定すると、レコードのソースファイルの名前がレコードに追加されます。パラメータ値は列名として使用されます。 + `attachTimestamp` — 列名として使用する適切な形式の文字列。このオプションを指定すると、レコードのソースファイルの変更時刻がレコードに追加されます。パラメータ値は列名として使用されます。