AWS Glue でのストリーミング ETL ジョブ

• Avro ストリーミングソースに関する注意事項と制約事項

• ストリーミングソースへの grok パターンの適用

Amazon Kinesis Data Streams のストリーミング ETL プロパティを設定するには (AWS Glue API または AWS CLI)

• 同じアカウント内の Kinesis ソースのストリーミング ETL プロパティを設定するには、CreateTable API オペレーションまたは create_table CLI コマンドの StorageDescriptor データ構造の streamName および endpointUrl パラメータを指定します。

  "StorageDescriptor": {
  	"Parameters": {
  		"typeOfData": "kinesis",
  		"streamName": "sample-stream",
  		"endpointUrl": "https://kinesis.us-east-1.amazonaws.com"
  	}
  	...
  }
  					

  または、streamARN を指定します。

  "StorageDescriptor": {
  	"Parameters": {
  		"typeOfData": "kinesis",
  		"streamARN": "arn:aws:kinesis:us-east-1:123456789:stream/sample-stream"
  	}
  	...
  }

• 別のアカウントの Kinesis ソースのストリーミング ETL プロパティを設定するには、CreateTable API オペレーションまたは create_table CLI コマンドの StorageDescriptor データ構造の streamARN、awsSTSRoleARN、および awsSTSSessionName (オプション) パラメータを指定します。

  "StorageDescriptor": {
  	"Parameters": {
  		"typeOfData": "kinesis",
  		"streamARN": "arn:aws:kinesis:us-east-1:123456789:stream/sample-stream",
  		"awsSTSRoleARN": "arn:aws:iam::123456789:role/sample-assume-role-arn",
  		"awsSTSSessionName": "optional-session"
  	}
  	...
  }

• スキーマの検出が有効になっている場合、Avro スキーマをペイロードに含める必要があります。無効にすると、ペイロードにはデータのみが含まれます。

• 一部の Avro データ型は、ダイナミックフレームでサポートされていません。AWS Glue コンソールの [create table] (テーブル作成) ウィザードの [Define a schema] (スキーマの定義) でスキーマを定義するときに、これらのデータ型を指定することはできません。スキーマの検出中に、Avro スキーマでサポートされていない型は、次のようにサポートされている型に変換されます。

  • EnumType => StringType

  • FixedType => BinaryType

  • UnionType => StructType

• コンソールの [Define a schema] (スキーマの定義) ページでテーブルスキーマを定義する場合は、スキーマの暗黙のルート要素タイプは record です。record 以外のルート要素タイプ (array や map など) が必要な場合、[Define a schema] (スキーマの定義) ページでスキーマを指定することはできません。代わりに、そのページをスキップして、スキーマをテーブルプロパティとして、または ETL スクリプト内で指定する必要があります。

  • テーブルのプロパティでスキーマを指定するには、テーブル作成ウィザードを完了し、テーブルの詳細を編集し、[Table properties] (テーブルのプロパティ) に新しいキーと値のペアを追加します。avroSchema キーを使用して、次のスクリーンショットに示すように、値のスキーマ JSON オブジェクトを入力します。

    

  • ETL スクリプトでスキーマを指定するには、次の Python と Scala の例に示すように、datasource0 代入ステートメントを修正し、avroSchema キーを additional_options 引数に追加します。

    Python

    SCHEMA_STRING = ‘{"type":"array","items":"string"}’
    datasource0 = glueContext.create_data_frame.from_catalog(database = "database", table_name = "table_name", transformation_ctx = "datasource0", additional_options = {"startingPosition": "TRIM_HORIZON", "inferSchema": "false", "avroSchema": SCHEMA_STRING})
    

    Scala

    val SCHEMA_STRING = """{"type":"array","items":"string"}"""
    val datasource0 = glueContext.getCatalogSource(database = "database", tableName = "table_name", redshiftTmpDir = "", transformationContext = "datasource0", additionalOptions = JsonOptions(s"""{"startingPosition": "TRIM_HORIZON", "inferSchema": "false", "avroSchema":"$SCHEMA_STRING"}""")).getDataFrame()
    

• AWS Glue ストリーミング ETL ジョブでの自動解凍は、サポートされた圧縮タイプに対してのみ使用できます。また、以下の点にも注意してください。

  • Framed Snappy とは、公式のフレーミング形式を使用する Snappy を指します。

  • Deflate がサポートされるのは、Glue バージョン 3.0 です (Glue バージョン 2.0 ではサポートされません)。

• スキーマ検出を使用する場合、ストリーミングデータの結合を実行できません。

• AWS Glue のストリーミング ETL ジョブは、Avro 形式の AWS Glue スキーマレジストリに対する Union データ型をサポートしません。

• ETL スクリプトでは、AWS Glue の組み込み変換と Apache Spark 構造化ストリーミングのネイティブな変換を使用できます。詳細については、Apache Spark ウェブサイトの「ストリーミング DataFrames/Datasets の操作」または「AWS Glue PySpark 変換リファレンス」を参照してください。

• AWS Glue ストリーミング ETL ジョブでは、チェックポイントを使用して、読み取ったデータの追跡が行われます。このため、停止して再起動されたジョブは、ストリーム内で中断した位置から開始されます。データを再処理する場合は、スクリプト内で参照されているチェックポイントフォルダを削除することができます。

• ジョブのブックマークはサポートされていません。

• ジョブで Kinesis Data Streams の拡張ファンアウト機能を使用するには、「Kinesis ストリーミングジョブでの拡張ファンアウトの使用」を参照してください。

• AWS Glue スキーマレジストリ から作成されたData Catalog テーブルを使用する場合、新しいスキーマバージョンが使用可能になったら、新しいスキーマを反映するには、次の操作を行う必要があります。

  1. テーブルに関連するジョブを停止します。

  2. Data Catalog テーブルのスキーマを更新します。

  3. テーブルに関連付するジョブを再起動します。