AWS Glue Streaming

AWS Glue のコンポーネントである AWS Glue Streaming を使用すると、ストリーミングデータをほぼリアルタイムで効率的に処理できるようになり、データインジェスト、処理、機械学習などの重要なタスクを実行できるようになります。AWS Glue Streaming は、Apache Spark Streaming フレームワークを使用して、ストリーミングデータを大規模に処理できるサーバーレスサービスを提供します。AWS Glue は、サーバーレスインフラストラクチャ、自動スケーリング、ビジュアルジョブ開発、ストリーミングジョブの瞬時オンノートブック、その他のパフォーマンスの向上など、Apache Spark 上でさまざまな最適化を提供します。

ストリーミングのユースケース

AWS Glue Streaming の一般的なユースケースには以下が含まれます。

ほぼリアルタイムのデータ処理: AWS Glue Streaming により、組織はストリーミングデータをほぼリアルタイムで処理できるため、最新の情報に基づいて洞察を導き出し、タイムリーな意思決定を行うことができます。

不正検出: AWS Glue Streaming はストリーミングデータのリアルタイム分析に利用できるため、クレジットカード詐欺、ネットワーク侵入、オンライン詐欺などの不正行為の検出に役立ちます。受信データを継続的に処理して分析することで、疑わしいパターンや異常を迅速に特定できます。

ソーシャルメディア分析: AWS Glue Streaming では、ツイート、投稿、コメントなどのソーシャルメディアデータをリアルタイムで処理できるため、組織はトレンドの監視、感情分析、ブランドの評判の管理をリアルタイムで行うことができます。

モノのインターネット (IoT) 分析: AWS Glue Streaming は、IoT デバイス、センサー、接続された機械によって生成される高速データストリームの処理と分析に適しています。これにより、リアルタイムの監視、異常検知、予知保全、およびその他の IoT 分析のユースケースが可能になります。

クリックストリーム分析: AWS Glue Streaming では、ウェブサイトやモバイルアプリケーションからのリアルタイムのクリックストリームデータを処理して分析できます。これにより、企業はユーザーの行動に関する洞察を得たり、ユーザー体験をパーソナライズしたり、リアルタイムのクリックストリームデータに基づいてマーケティングキャンペーンを最適化したりすることができます。

ログの監視と分析: AWS Glue Streaming では、サーバー、アプリケーション、ネットワークデバイスからのログデータをリアルタイムで継続的に処理および分析できます。これは、異常の検出、問題のトラブルシューティング、システムの状態とパフォーマンスの監視に役立ちます。

レコメンデーションシステム: AWS Glue Streaming では、ユーザーアクティビティデータをリアルタイムで処理し、レコメンデーションモデルを動的に更新できます。これにより、ユーザーの行動や好みに基づいた、パーソナライズされたリアルタイムのレコメンデーションが可能になります。

これらは、AWS Glue Streaming を適用できるさまざまなユースケースの例です。AWS エコシステムやマネージドサービスとの統合により、クラウドでのリアルタイムのストリーム処理や分析を便利に行えるようになります。

AWS Glue Streaming を使用することの利点は何ですか。

AWS Glue Streaming を使用する利点は以下のとおりです。

AWS Glue Streaming はどのようなときに使うべきですか。

ストリーミングのユースケースに関して言えば、多くの選択肢があります。AWS Glue Streaming は以下のシナリオで推奨されます。

サポートされているデータソース

AWS Glue Streaming では、次のデータソースがサポートされています。

サポートされるデータターゲット

AWS Glue Streaming では、次のようなさまざまなデータターゲットがサポートされています。

ストリーミングジョブのリアルタイムモードの有効化

リアルタイムモード (RTM) は、AWS Glue 6.0 で利用可能な Spark 構造化ストリーミングの新しい実行モデルです。RTM はエンドツーエンドのレイテンシーを秒単位または分単位から秒未満に短縮します。リアルタイムモードは、Spark 構造化ストリーミングジョブにのみ適用されます。レガシー Spark ストリーミング (DStreams) やその他のジョブタイプには適用されません。

RTM は Trigger.RealTime を使用します。タスクはバッチ期間 (デフォルトは 5 分) 内に継続的に実行され、間隔をまたいでデータを蓄積するのではなく、到着時にレコードを処理します。これは、forEachBatch/Trigger.ProcessingTime が各間隔でタスクをポーリング、処理、コミット、再起動するデフォルトのマイクロバッチモデルとは異なります。

前提条件

リアルタイムモードを有効にする前に、ジョブが以下の要件を満たしていることを確認します。

リアルタイムモードを使用するタイミング

リアルタイムモードは、特定のクラスのストリーミングワークロード用に設計されています。以下の場合は、リアルタイムモードの使用を検討してください。

以下の場合は、引き続きマイクロバッチモードを使用します。

リアルタイムモードの仕組み

マイクロバッチモデルとリアルタイムモードの違いを以下に示します。

どちらのモードでも、同じチェックポイント形式と復旧メカニズムを使用します。主な違いは、タスクの有効期間です。マイクロバッチモードは、間隔ごとにタスクを終了し、再起動します。リアルタイムモードは、より長期間のバッチウィンドウでタスクを継続的に実行します。

リアルタイムモードを有効にするには

リアルタイムモードを有効にするには、--enable-real-time-mode ジョブ引数を true に設定します。この引数は、AWS Glue コンソールまたは API を使用して設定できます。

リアルタイムモードを有効にするには (コンソール)

リアルタイムモードを有効にするには (API)

--enable-real-time-mode フラグは、ジョブ定義の DefaultArguments マップに保存されます。ジョブを作成または更新するときに設定できます。

新しいジョブを作成するには (AWS CLI)

次のコマンドを実行します。

aws glue create-job \
  --name my-rtm-job \
  --role arn:aws:iam::123456789012:role/MyGlueRole \
  --glue-version 6.0 \
  --worker-type G.1X --number-of-workers 4 \
  --command '{"Name":"gluestreaming","ScriptLocation":"s3://my-bucket/scripts/rtm-job.scala"}' \
  --default-arguments '{
      "--enable-real-time-mode": "true",
      "--job-language": "scala",
      "--class": "GlueApp",
      "--TempDir": "s3://my-bucket/tmp/"
  }' \
  --region us-east-2

新しいジョブを作成するには (boto3)

以下のコードを使用します。

import boto3

glue = boto3.client("glue", region_name="us-east-2")

glue.create_job(
    Name="my-rtm-job",
    Role="arn:aws:iam::123456789012:role/MyGlueRole",
    GlueVersion="6.0",
    WorkerType="G.1X",
    NumberOfWorkers=4,
    Command={
        "Name": "gluestreaming",
        "ScriptLocation": "s3://my-bucket/scripts/rtm-job.scala",
    },
    DefaultArguments={
        "--enable-real-time-mode": "true",
        "--job-language": "scala",
        "--class": "GlueApp",
        "--TempDir": "s3://my-bucket/tmp/",
    },
)

既存のジョブを更新するには (AWS CLI)

次のコマンドを実行します。

aws glue update-job \
  --job-name my-existing-job \
  --job-update '{
      "GlueVersion": "6.0",
      "DefaultArguments": {
          "--enable-real-time-mode": "true",
          "--job-language": "scala"
      }
  }'

ストリーミングスクリプトの記述

ジョブ引数は、リアルタイムモードを使用する意図を宣言します。スクリプトがトリガーを選択します。

次の Scala の例は、Trigger.RealTime を使用するストリーミングクエリを示しています。

import org.apache.spark.sql.streaming.Trigger

val query = df.writeStream
  .format("kafka")
  .outputMode("update")
  .trigger(Trigger.RealTime(60000L))  // checkpoint interval in milliseconds
  .start()

query.awaitTermination()

Trigger.RealTime は、チェックポイント間隔をミリ秒単位で受け取ります。更新出力モードが必要です。追加モードでは OUTPUT_MODE_NOT_SUPPORTED がスローされます。

フラグが設定されている限り、モードを 1 つのスクリプトに混在させることができます。

dfA.writeStream.outputMode("update").trigger(Trigger.RealTime(60000L)).start()
dfB.writeStream.outputMode("append").trigger(Trigger.ProcessingTime("30 seconds")).start()

フラグがない場合の動作

--enable-real-time-mode フラグが設定されていない場合のジョブの動作を以下に示します。

考慮事項と制限事項

リアルタイムモードを使用する場合は、以下を考慮してください。