AWSで実現する大規模データ保存・分散処理 ―IcebergとSparkの仕組みと実践― Part2

AWSで実現する大規模データ保存・分散処理 ― IcebergとSparkの仕組みと実践 ― Part 2 山崎 拓也

山崎 拓也 所属: SIer 仕事: • AWS案件のアプリやインフラのリード • 社内AWSサポート 好き: 低レイヤ、SF、AWS AWSアワード: • 2026 AWS Community Builder (Serverless) • 2024～2025 Japan AWS Top Engineer • 2022～2025 Japan All AWS Certifications Engineer

アジェンダ Part1 • 要件とアーキテクチャ • Icebergの仕組み • S3 Tablesの実践的なポイント Part2 • Sparkの仕組み • Glue Spark Jobの実践的なポイント • まとめ

アーキテクチャ再掲 （仮想要件等はPart1を参照）

Sparkの仕組み

Apache Spark とは • 大規模データを高速に処理するための分散処理エンジン • Driverノード1つ、Workerノード複数 • Driverがデータ処理の実行計画を立て、ExecutorへTaskとして配布する • データをSparkパーティションとして分割し、複数ノードで並列処理する

Actionが実行されるまで、データロードや処理は実行されない • Transformation と Action の2種類のメソッドがある • Transformationメソッドは実行計画の作成のみ行う（遅延評価） • map() • filter() • groupByKey() • repartition() • Actionメソッドで実行計画が初めて実行される • reduce() • collect() • count() • save()

遅延評価をコードで確認 ❶ ❶ select/filterを行い、DataFrame定義しても、 まだデータはロードされない ❷ ❸ ❷ groupBy/agg/orderByをしても、 まだ実行されない ❸ show()により、 ここで初めて実際のロード・変換・集計が実行。 各ExecutorにTask配布して分散実行される

データの分散単位はSparkパーティション • ロード時、データは複数の Spark パーティションへ分散される • ロード時のパーティション配置は処理内容に最適化されているとは限らない • 処理パターンに合わせて、repartition() によりデータを再分散する repartition("device_id") 実際は偏りをなくすため パーティション数を指定し、 ハッシュ関数で均等に分散させる repartition(40, "device_id")

パフォーマンスに影響する注意が必要な操作 • WorkerからDriverに全てのデータを集める • collect() • toPandas() • シャッフル。ノード間やパーティション間でデータを再分配する操作 • groupByKey() • orderBy() • repartition() 特にシャッフルは生じやすいので注意

再利用するDataFrameはキャッシュに乗せる • キャッシュしない限り、アクションごとにDataFrameが再計算される • df.cache()：アクション実行後、Executorのキャッシュに乗せる • df.unpersist()：キャッシュから降ろす

パーティションごとに任意の関数で分散処理する • df.rdd.mapPartitions(func) • パーティションのレコードをループしたい場合 • func()にはパーティション内レコードのiteratorが渡される • df.mapInPandas(func, schema=...) • パーティションをPandas DataFrameとして扱いたい場合 • func()にはPandas DataFrameのiteratorが渡される

Glue Spark Jobの実践的なポイント

Workerノードのオートスケーリング設定ができる • オートスケーリングしない場合、ノードが余るとコストが余計にかかる オートスケーリング 料金 無効（デフォルト） 設定したワーカー数 × ワーカータイプのDPU × 実行時間 有効 実際に使用したDPU • ConnectionでVPC接続する場合、ノード毎にENI作成されるためIP数に注意 Worker type specifications table： https://docs.aws.amazon.com/glue/latest/dg/work er-types.html#worker-type-specifications

• https://docs.aws.amazon.com/glue/latest/dg/worker-types.html#worker-type-specifications

CloudWatchログはDriver、Executorごとにストリームが分かれる • 基本的にログは全て /aws-glue/jobs/error ロググループに出力（変更可） • Workerノードごとに必ず１つのExecutor。本当に便利 Driver Executor Workerノード数 と等しい • 各TaskのパーティションIDなどの情報はTaskContextから取得できる Worker type specifications table：https://docs.aws.amazon.com/glue/latest/dg/worker-types.html#worker-type-specifications

• https://docs.aws.amazon.com/glue/latest/dg/worker-types.html#worker-type-specifications

コンソールのSpark UIから詳細なパフォーマンスが確認できる • タスク数やどの処理に時間がかかっているのかなど

Icebergパーティション通りのSparkパーティションになる保証はない • 処理パターンに最適化するにはrepartition()が必要 • 検証時、コンパクション前は同一device_idがパーティションを跨いでロー ドされた • コンパクション後は同一device_idが同一パーティションに綺麗にロードさ れたが、保証はない • パーティション分割数は、Executorの総CPUコア数より多めに設定し、継続 的にタスクが割り当たるようにする Worker type specifications table： https://docs.aws.amazon.com/glue/latest/dg/work er-types.html#worker-type-specifications

• https://docs.aws.amazon.com/glue/latest/dg/worker-types.html#worker-type-specifications

Redshiftへの書き込みもExecutorを使って効率よく行う • Driverに集めない • 処理後データは各ExecutorからS3へ出力される • Driverからの1度のCOPYクエリ実行でRedshiftへ書き込まれる • S3の出力ファイルは自動削除されないためライフサイクルで消す必要あり

デバイスID毎、イベント時間順に分散データ処理する際の例 パーティション分割、 ソート、複雑な処理 処理日時列を追加 save() がアクション

まとめ • IcebergもSparkも、仕組みを理解して上手く使うことが大切 • S3 Tablesへの挿入はFirehoseを使って料金と効率を最適化 • Glue Spark Jobはパーティションとアクション・シャッフルを意識する

ご清聴ありがとうございました。