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

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

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

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

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

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オプションGetObjectAPIを に提供することで、S3 オブジェクトから分割を取得できます。次の図を参照して、これが実際にどのように機能するかを確認してください。

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

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

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

ファイルレイアウト – S3 データを、データの使用方法に基づいて異なるパスのオブジェクトに保存することで、関連するデータを効率的に取得できます。詳細については、Amazon S3ドキュメント」の「プレフィックスを使用してオブジェクトを整理する」を参照してください。 AWS Glue は、キーと値を 形式で Amazon S3 プレフィックスに保存しkey=value、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 パーティションインデックスを使用して Data Catalog からパーティションメタデータを取得する際のレイテンシーを短縮することを検討してください。

には 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"}
)

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

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

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