DEA / ドメイン1 データの取り込みと変換
変換処理のトラブルシューティングと高速化
や の のジョブは、データが増えたり偏ったりすると、急に遅くなったり、メモリ不足で失敗したりします。DEA では、症状 (メモリ不足、1つのタスクだけ遅い、出力のファイルが多すぎる) から原因を推定し、ワーカーの増強、パーティション数の調整、結合の方法の変更、コードの書き方の改善のどれを選ぶかが問われます。
Point
- シャッフル (shuffle) は、結合や集計のためにデータをキーごとにワーカー間で移し替えることです。ネットワークとディスクを大量に使うため、遅さの主な原因になります。
- データスキュー (data skew) は、特定のキーにデータが偏り、一部のタスクだけが極端に重くなることです。
- OOM = Out Of Memory (メモリ不足) です。
- 和訳
- Skew: ゆがみ、偏り
- Weblio英和辞書「skew」
概要
調べる道具は次のとおりです。
| 道具 | わかること |
|---|---|
| UI (、) | ステージとタスクの所要時間、の量、タスクごとの読み込み量の偏り |
| Glue のジョブ・オブザーバビリティメトリクス () | ドライバーとのメモリとディスク、ワーカーの利用率、 (glue.driver.skewness.stage / job)、エラーの分類 (OUT_OF_MEMORY、PERMISSION、THROTTLING など) |
| CloudWatch Logs (継続的なログ記録) | 実行中のエラーとスタックトレース |
| の EXPLAIN / クエリの統計 | スキャンしたパーティション、各段階の時間 |
主要技術
①メモリ不足 (OOM) ⚠️ 頻出
| 発生場所 | よくある原因 | 対処 |
|---|---|---|
| ドライバー | collect() や toPandas() で全データをドライバーへ集める。数十万のの一覧を作る | collect をやめて分散したまま書き出す。groupFiles で読み取りをまとめる。 のファイル一覧の取得を分散させる設定を使う |
| 1つのパーティションが大きすぎる。で1つのキーにデータが集中。大きなテーブル同士の結合 | パーティション数を増やす (repartition)。スキューの対処。大きいワーカー (G.2X 以上) や R 系のワーカー | |
| ディスク | のデータがワーカーのディスクに収まらない | ワーカーを大きくするか増やす。シャッフルの量を減らす |
ワーカーを大きくすれば多くの場合は動きますが、費用が増えます。原因がスキューや collect にあるなら、ワーカーを大きくしても解決しないか、無駄が大きくなります。
⚠️ 頻出: 「ドライバーのメモリ不足で失敗。入力は S3 の数十万個の小さな 」は、小さいファイルの問題です。groupFiles / groupSize で読み取りをまとめるか、ファイルを結合します。
②データスキュー ⚠️ 頻出
UI で、同じステージのタスクのうち一部だけが他よりずっと長くかかり、読み込み量も大きい場合はです。 では glue.driver.skewness.stage の値でも確かめられます。
| 対処 | 内容 |
|---|---|
| (salting) | 偏ったキーにランダムな接尾辞 (例: キー_0〜キー_9) を付けて分散させ、集計してから接尾辞を外して再集計する |
| 適応型クエリ実行 (AQE) | Spark 3 の機能。実行中の統計を見て、偏った結合のパーティションを自動で分割する (skew join の最適化) |
| 片方が小さければ、せずに全ワーカーへ配る (偏ったキーのシャッフルを避けられる) | |
| 偏った値を分けて処理 | NULL や「不明」のような値に偏るなら、それだけを別に処理する |
スキューの検出とサンプリングの考え方は、「データサンプリングとデータスキュー」の章でも扱います。
③シャッフルを減らす結合と集計
| 手段 | 内容 |
|---|---|
小さいテーブル (マスターなど) を全ワーカーへ配り、大きいテーブルをせずに結合する (broadcast() のヒント) | |
| 早めの絞り込み | 結合の前に filter と列の選択を行い、移動するデータを減らす |
| 述語と列のプッシュダウン | やカタログのパーティションで、読む段階で減らす ( の) |
| シャッフルのパーティション数 | spark.sql.shuffle.partitions を、データ量に合った値にする (AQE は自動で結合して調整する) |
| 同じキーでの事前のパーティション分割・化 | 繰り返し同じキーで結合するなら、保存時にそろえておく |
小さい商品マスターをブロードキャストして、大きな注文テーブルと結合する例です。
from pyspark.sql.functions import broadcast
result = orders.join(broadcast(products), "product_id")
④パーティション数とファイル数
| 状況 | 対処 |
|---|---|
| 入力のファイルが小さく大量 | の groupFiles=‘inPartition’ と groupSize でまとめて読む |
| 出力がだらけ | 書き込みの前に coalesce (減らすだけ、なし) か repartition (シャッフルして均等に) で数を調整 |
| パーティション数が少なく並列にならない | repartition で増やす。ワーカー数とコア数に見合う数にする |
| 出力のパーティション列の値が多すぎる | パーティションのキーを見直す (日付の粒度を日単位にするなど) |
⑤ランタイムを短くするコードの最適化
| 書き方 | 理由 |
|---|---|
| Python の UDF より の組み込み関数 | 組み込み関数は最適化され、Python との間のデータの受け渡しがない |
| 必要な列だけを select する | 読む量との量が減る |
| 何度も使う DataFrame は cache / persist | 同じ計算を繰り返さない (使い終われば解放する) |
| collect、count の多用を避ける | 不要なアクションのたびに処理が最初から走る |
| 増分の処理 (ブックマーク) | 毎回全量を処理しない |
| 列指向の形式と圧縮で入力を持つ | 読み込みが速い |
⑥よくある変換エラー ⚠️ 頻出
| エラー・症状 | 主な原因 | 対処 |
|---|---|---|
| AccessDenied () | ジョブの ロールに権限がない、ポリシー、 のキーの権限 | ロールのポリシー、キーポリシーに kms:Decrypt などを追加 |
| 接続のタイムアウト () | の経路、、S3 のがない | 「データソースへの接続」の章の切り分け |
| 型の不一致、列が NULL になる | 一部のファイルで型が違う、スキーマの変化 | ResolveChoice、ApplyMapping、スキーマの明示 |
| 文字化け | 文字コードの違い (Shift_JIS など) | 読み込み時に文字コードを指定する |
| 同じデータが二重に入る | ブックマークの無効化や巻き戻し、再実行がでない | ブックマークを有効にする、上書きや で書き込む |
| ThrottlingException | 書き込み先 ( など) や の上限 | 書き込みの速さを抑える、再試行、容量を上げる |
| ジョブが開始できない ( 不足) | VPC のの空きアドレスが足りない | 大きなサブネット、ジョブ実行のキューイング |
試験対策
①判断マトリクス
| 問題文のキーワード | 選ぶ対処 |
|---|---|
| 一部のタスクだけが極端に遅い | → 、AQE |
| 小さいマスターと大きいテーブルの結合が遅い | |
| ドライバーが OOM、入力は大量の | groupFiles でまとめる、ファイルの結合 |
| ドライバーが OOM、collect を使っている | collect をやめて分散したまま書く |
| が OOM | パーティションを増やす、大きいワーカー / R 系 |
| 出力のファイルが小さく大量 | coalesce / repartition |
| 必要なパーティションだけ読みたい | |
| どのステージが遅いかを調べる | UI |
| メモリや偏りの推移を で | のオブザーバビリティ |
| 実行中のエラーをすぐに見たい | 継続的なログ記録 |
| は読めるが で暗号化されたオブジェクトだけ失敗 | KMS のキーポリシーとロールの権限 |
| Python の UDF が遅い | 組み込み関数に書き換える |
②よくある誤答
- どんな遅さにもワーカーを増やすこと。は1つのタスクに集中するため、ワーカーを増やしても速くなりません。
- 出力のファイル数を減らすのに、毎回 repartition(1) にすること。1つのワーカーに集中して遅く、メモリ不足の原因になります。
- 失敗の原因を調べる前にジョブを再試行し続けること。権限や設定の誤りは何度再試行しても失敗します。