データスキャンの量を減らす

ファイルサイズ – 入力と出力のファイルサイズを中程度の範囲 (128 MB など) に維持することをお勧めします。ファイルが小さすぎたり、大きすぎたりすると、問題が発生する可能性があります。

小さなファイルが多数あると、次の問題が発生します。

一方、ファイルタイプが分割可能でなく (gzip など）、ファイルが大きすぎる場合、Spark アプリケーションは 1 つのタスクでファイル全体の読み取りが完了するまで待つ必要があります。

小さなファイルごとに Apache Spark タスクを作成するときに発生する過剰な並列処理を減らすには、DynamicFrames のファイルグループを使用します。このアプローチにより、Spark ドライバーから OOM 例外が発生する可能性が低くなります。ファイルグループ化を設定するには、 パラメータgroupFilesと groupSizeパラメータを設定します。次のコード例では、これらのパラメータを含む ETL スクリプトで AWS Glue DynamicFrame API を使用しています。

dyf = glueContext.create_dynamic_frame_from_options("s3",
    {'paths': ["s3://input-s3-path/"],
    'recurse':True,
    'groupFiles': 'inPartition',
    'groupSize': '1048576'},
    format="json")

圧縮 — S3 オブジェクトが数百メガバイト内にある場合は、圧縮することを検討してください。さまざまな圧縮形式があり、2 つのタイプに大別できます。

Spark (およびその他の一般的な分散処理エンジン) では、ソースデータファイルをエンジンが並行して処理できるチャンクに分割します。これらの単位は分割と呼ばれることがよくあります。データが分割可能な形式になると、最適化された AWS Glue リーダーは、特定のブロックのみを取得するRangeオプションを GetObject API に提供することで、S3 オブジェクトから分割を取得できます。これが実際にどのように機能するかを確認するには、次の図を検討してください。

圧縮されたデータは、ファイルが最適なサイズであるか、ファイルが分割可能である限り、アプリケーションを大幅に高速化できます。データサイズが小さいほど、Amazon S3 からスキャンされるデータとAmazon S3 から Spark クラスターへのネットワークトラフィックが減少します。一方、データを圧縮および解凍するには、より多くの CPU が必要です。必要なコンピューティングの量は、圧縮アルゴリズムの圧縮率でスケールされます。分割可能な圧縮形式を選択するときは、このトレードオフを考慮してください。

ファイル形式 – 列形式を使用します。Apache Parquet と Apache ORC は一般的な列データ形式です。Parquet と ORC は、列ベースの圧縮、データ型に基づく各列のエンコードと圧縮を採用することで、データを効率的に保存します。Parquet エンコーディングの詳細については、「Parquet エンコーディング定義」を参照してください。Parquet ファイルも分割可能です。

Columnar は、グループ値を列ごとにフォーマットし、ブロックにまとめて保存します。列形式を使用する場合、使用しない列に対応するデータのブロックをスキップできます。Spark アプリケーションは、必要な列のみを取得できます。一般的に、圧縮率やデータブロックをスキップすると、Amazon S3 から読み取るバイト数が少なくなり、パフォーマンスが向上します。どちらの形式も、I/O を減らすための以下のプッシュダウンアプローチをサポートしています。

ファイルレイアウト – データの使用方法に基づいて S3 データを異なるパスのオブジェクトに保存することで、関連するデータを効率的に取得できます。詳細については、「Amazon S3 ドキュメント」の「プレフィックスを使用してオブジェクトを整理する」を参照してください。 AWS Glue は、キーと値を 形式で Amazon S3 プレフィックスに保存しkey=value、データを Amazon S3 パスでパーティション化することをサポートしています。 Amazon S3 データをパーティション化することで、各ダウンストリーム分析アプリケーションでスキャンされるデータ量を制限し、パフォーマンスを向上させ、コストを削減できます。詳細については、「 での ETL 出力のパーティションの管理 AWS Glue」を参照してください。

パーティション分割はテーブルをさまざまな部分に分割し、次の例に示すように、年、月、日などの列値に基づいて関連するデータをグループ化されたファイルに保持します。

# Partitioning by /YYYY/MM/DD
s3://<YourBucket>/year=2023/month=03/day=31/0000.gz
s3://<YourBucket>/year=2023/month=03/day=01/0000.gz
s3://<YourBucket>/year=2023/month=03/day=02/0000.gz
s3://<YourBucket>/year=2023/month=03/day=03/0000.gz
...

データセットのパーティションは、 のテーブルでモデリングすることで定義できます AWS Glue Data Catalog。その後、次のようにパーティションプルーニングを使用してデータスキャンの量を制限できます。

Amazon S3 パーティションが多すぎる – 数千の値を持つ ID 列など、幅広い値を含む列で Amazon S3 データをパーティション化することは避けてください。これにより、バケット内のパーティションの数が大幅に増加する可能性があります。可能なパーティションの数は、パーティション分割したすべてのフィールドの積であるためです。パーティションが多すぎると、以下が発生する可能性があります。

たとえば、 partitionByまたは で日付タイプを設定する場合partitionKeys、 などの日付レベルのパーティショニングyyyy/mm/ddは多くのユースケースに適しています。ただし、 yyyy/mm/dd/<ID>は多数のパーティションを生成するため、パフォーマンス全体に悪影響を与える可能性があります。

一方、リアルタイム処理アプリケーションなどの一部のユースケースでは、 などの多くのパーティションが必要ですyyyy/mm/dd/hh。ユースケースで大量のパーティションが必要な場合は、AWS Glue パーティションインデックスを使用して、データカタログからパーティションメタデータを取得するためのレイテンシーを短縮することを検討してください。

DynamicFrame の場合は、sampleQuery オプションを使用します。を使用する場合はcreate_dynamic_frame.from_catalog、次のように引additional_options数を設定します。

query = "SELECT * FROM <TableName> where id = 'XX' AND"
datasource0 = glueContext.create_dynamic_frame.from_catalog(
  database = db,
  table_name = table,
  additional_options={
      "sampleQuery": query,
      "hashexpression": key,
      "hashpartitions": 10,
      "enablePartitioningForSampleQuery": True
    },
  transformation_ctx = "datasource0"
)

の場合using create_dynamic_frame.from_options、引connection_options数を次のように設定します。

query = "SELECT * FROM <TableName> where id = 'XX' AND"
datasource0 = glueContext.create_dynamic_frame.from_options(
    connection_type = connection, 
    connection_options={
        "url": url,
        "user": user,
        "password": password,
        "dbtable": table,
        "sampleQuery": query,
        "hashexpression": key,
        "hashpartitions": 10,
        "enablePartitioningForSampleQuery": True
    }
)

DataFrame の場合は、 クエリオプションを使用します。

query = "SELECT * FROM <TableName> where id = 'XX'"
jdbcDF = spark.read \
.format('jdbc') \
.option('url', url) \
.option('user', user) \
.option('password', pwd) \
.option('query', query) \
.load()

Amazon Redshift の場合は、 AWS Glue 4.0 以降を使用して Amazon Redshift Spark コネクタのプッシュダウンサポートを活用します。

dyf = glueContext.create_dynamic_frame.from_catalog(
    database = "redshift-dc-database-name", 
    table_name = "redshift-table-name", 
    redshift_tmp_dir = args["temp-s3-dir"], 
    additional_options = {"aws_iam_role": "arn:aws:iam::role-account-id:role/rs-role-name"}
)

他のデータベースについては、そのデータベースのドキュメントを参照してください。

すべての連続ジョブ実行の完全なスキャンを回避し、前回のジョブ実行時に存在しなかったデータのみを処理するには、ジョブのブックマークを有効にします。

処理する入力データの量を制限するには、ジョブブックマークを使用して制限付き実行を有効にします。これにより、ジョブ実行ごとにスキャンされるデータの量を減らすことができます。