Dataflow周りで地雷を踏みまくった話

Scio + Dataflowでバッチを実装した際に色々な問題に突き当たったときの話

Transcript

1. Dataflow周りで地雷を踏みま くった話 2020/07 @mikesorae
2. 背景 • 業務の一環でデータ分析レポートを作っている • 1000行〜3000行くらいのSQLをいつも書いている => つらい • 条件文をRubyで組み立てて最終結果もRubyで加工している • 一ヶ月も経つと何やってるか読めない => つらい • window関数を多用するためデータ抽出が1回8時間〜24時間かかる => つらい • SQLをこね回しているので仕様変更/デバッグ/レビューも時間がかかる => つらい メンテナンス性とデータ処理時間改善のためBQ + Google Cloud Dataflowへ
3. Google Cloud Dataflowとは • Apache Beam Pipelineを実行するフルマネージドな大規模データ処理基盤 • 計算リソースを自動でプロビジョニング、オートスケール
4. Apache Beamとは • Batch処理とStreaming処理の統合プログラミングモデル ◦ Batch + Stream = Beam • Dataflow用のプログラミングモデルだったが、Googleによって2016年にOSS化された • Dataflow以外にもSpark、Flink、Nemo、Samzaを実行環境として利用できる • JavaとPythonとGo版のSDKを提供している ◦ Go版はDataflowで動くのか未確認
5. Apache Beamのプログラミングモデル(1) • Pipeline ◦ データの読み込みから出力まで含めた一連のデータ処理タスクを包括する概念 • PCollection ◦ データ処理の対象となる分散データセット ◦ テキストファイルのような有限データも、 Pub/Subのような無限データも扱える • PTransform ◦ Pipeline中の各ステップのデータ処理操作 ◦ 1〜複数個のPCollectionを受け取って0〜1個のPCollectionを返す
6. Apache Beamのプログラミングモデル(2) Input PTransform PTransform PCollection PCollection PTransform Output データの読み込み データの書き込み
7. WordCountの例 https://github.com/apache/beam/blob/master/examples/java/src/main/java/org/apache/beam/examples/MinimalWordCount.java Pipeline p = Pipeline.create(options); p.apply(TextIO.read().from("gs://apache-beam-samples/shakespeare/*")) .apply( FlatMapElements.into(TypeDescriptors.strings()) .via((String line) -> Arrays.asList(line.split("[^\\p{L}]+")))) .apply(Filter.by((String word) -> !word.isEmpty())) .apply(Count.perElement()) .apply( MapElements.into(TypeDescriptors.strings()) .via( (KV<String, Long> wordCount) -> wordCount.getKey() + ": " + wordCount.getValue())) .apply(TextIO.write().to("wordcounts"));
8. 検討した構成 最終的にScioを採用
9. Java + Apache Beam
10. 独自の処理関数の定義 // 文字数を数える処理の関数部 static class ComputeWordLengthFn extends DoFn<String, Integer> { @ProcessElement public void processElement(@Element String word, OutputReceiver<Integer> out) { // Use OutputReceiver.output to emit the output element. out.output(word.length()); } } Genericsやアノテーションなどで記述量がそこそこ多くなる JavaよりはKotlinの方が楽かと思い選定から除外
11. Kotlin + Apache Beam
12. JvmWildcardを付けないと実行時エラー typecheck時に required: Iterable<? extends MyValue> found: Iterable<MyValue> で実行時エラーになる。 解決方法: MyValueに@JvmWildCardをつけ る。 class MyFn : DoFn< KV<String, Iterable<MyValue>>, // InputT String // OutputT >() { ...
13. DefaultCoderアノテーションが効かない PCollectionに渡す値はSerializeする必要が ある。 Javaだと@DefaultCoderを付ければ自動で変 換してくれるがKotlinだと動かない。 SchemaBuilderでschemaを組み立て、自前で CodeRegistry.registerCoderForClass で coderを登録しても解決しなかった。 解決方法: PCollectionのsetCoderでcoderを 都度指定 @DefaultCoder(AvroCoder::class) data class Parson( var id: String? = null, var gender: String? = null, var age: String? = null )
14. Kotlin + Apache Beam • Tupleが使えない • 自作PairCoderを作ってみたけどJvmSuppresWildcardが必要 • lambdaの型チェックがうまくいかない(コンパイルは通るけど実行時に落ちる) • 結局型を自分で書かないといけない箇所が多い • 全体的にリフレクション周りでハマる • 期待していたよりもコード量が減らず、玄人作業になる部分が多かったため、採用を見送 り
15. Scala + Scio
16. Scioとは • Spotify製のScala版Apache Beamラッパー • Java版のApache Beamよりも処理をシンプルに記述できる
17. Apache BeamとScio • Pipeline => ScioContext • PCollection => SCollection • PTransform => .map / .flatMap / .filter / .reduce / .transform • PipelineResult => ScioResult • innerでApache Beamのインスタンスを取り出してApache Beamとして書くこともできる。 Kotlin/Java + Apache Beamよりも圧倒的に書きやすく、 型周りで特にハマることもなかったため採用
18. WordCountの例 https://github.com/spotify/scio/blob/master/scio-examples/src/main/scala/com/spotify/scio/examples/MinimalWordCount.scala sc.textFile(args.getOrElse("input", ExampleData.KING_LEAR)) // => SCollection[String] .transform("counter") { _.flatMap(_.split("[^a-zA-Z']+").filter(_.nonEmpty)) // => SColection[String] .countByValue // => SCollection[(String, Long)] } .map(t => t._1 + ": " + t._2) // => SCollection[String] .saveAsTextFile(args("output"))
19. ValueProviderの制約にひっかかる
20. Dataflowジョブの実行方式 • Dataflowのジョブの実行方式は大きく分けて2つある(※この後増える) ◦ Dataflow Runner ▪ ビルド/デプロイ/実行をワンセットで行う ▪ 毎回ビルドするので時間がかかる ◦ カスタムテンプレート ▪ ビルド/デプロイと実行を分けて行う ▪ 実行のたびにビルドを行わないので何度も実行するならこちらの方が良い ▪ 実行するためにGUIかDataflow APIを使う必要がある
21. カスタムテンプレートの制約 • カスタムテンプレートを実行する際はValueProviderを経由して値を受け取る必要がある • ValueProviderの値はPipelineの実行中にしか取得できない ◦ ValueProviderの値で動的にPipelineを組み立てることができない
22. ValueProvider // カスタムテンプレートを使わない場合 public interface MyOptions : DataflowPipelineOptions { @Validation.Required fun getMyValue(): String } // カスタムテンプレートを使う場合 public interface MyOptions : DataflowPipelineOptions { @Validation.Required fun getMyValue(): ValueProvider[String] }
23. やりたかったこと BQ データ取得 何らかの処理 data Output Output 何らかの処理 何らかの処理 data パラメータの個数に応じて書き 込み先を分岐
24. やりたかったこと BQ データ取得 何らかの処理 data Output Output 何らかの処理 何らかの処理 data 実行中にしか値が取れないので 分岐を組み立てられない
25. 解決方法 • Flex Template(beta版)を使う ◦ 自前でビルドしたDockerイメージをジョブに使うことができる ◦ 内部的にはローカルと同じ仕組みで動いているため、 Value Providerを使わなくても値が取得でき る ◦ 現在はJava版のみ対応
26. ScioのFat Jarが必要になる
27. Flex Templateの制約 • Dockerイメージのベースイメージは gcr.io/dataflow-templates-base/java8-template-launcher-base (またはjava11) にす る必要がある • 指定されたディレクトリにFat Jarを配置しなければならない ◦ 専用のlauncherから呼び出されており、class pathが指定できない ◦ ベースイメージのソースコードも公開されていない
28. ScioでFat Jarを作る • Apache BeamのmvnではデフォルトでFat Jarだが、Scioでは自作する必要がある • Scioの公式FAQでは、マージ戦略が複雑になるためsbt-assemblyではなくsbt-packか sbt-native-packagerを推奨している • 調べてみたところ、どちらもFat Jarではなくlibsにjarを配置するタイプだったため使えな かった • よくよくScioの公式のexampleを見るとbuild.sbtでsbt-assemblyを使っていた
29. 解決方法 • 公式のbuild.sbtにあるsbt-assemblyの設定を使う
30. sbt-assemblyの設定 https://github.com/spotify/scio/blob/master/build.sbt lazy val assemblySettings = Seq( test in assembly := {}, assemblyMergeStrategy in assembly ~= { old => { case s if s.endsWith(".properties") => MergeStrategy.filterDistinctLines case s if s.endsWith("public-suffix-list.txt") => MergeStrategy.filterDistinctLines case s if s.endsWith("pom.xml") => MergeStrategy.last case s if s.endsWith(".class") => MergeStrategy.last case s if s.endsWith(".proto") => MergeStrategy.last case s if s.endsWith("libjansi.jnilib") => MergeStrategy.last ….
31. systemdの起動パラメータ文字数に ひっかかる
32. Flex TemplateのジョブのTimeOut Error • Flex Templateは実行できるようになったが、時々いつまでたってもジョブが実行されず TimeOutになった • パラメータとして大量の文字列を渡していた • パラメータ文字列を削ると動いた ◦ 2500文字くらいが境界値ぽい
33. Flex Templateのパラメータ渡しの仕組み 1. Dataflow APIのlaunch_flex_templateにパラメータを渡して実行 2. GCPがlauncher EC2インスタンスを作成 3. EC2のカスタムメタデータでパラメータ文字列をlauncherインスタンスに渡す a. user-dataでdocker起動用のsystemdのconfが渡される b. template-container-argsでdockerコマンドの引数が渡される c. template-container-argsの引数がsystemdのExecStartに渡される 4. systemdがdockerコンテナを起動する
34. systemdの設定 #cloud-configs users: - name: cloudservice uid: 2000 write_files: - path: /etc/systemd/system/cloudservice.service permissions: 0644 owner: root content: | [Unit] Description=Start a dynamic template launcher docker container Wants=gcr-online.target After=gcr-online.target [Service] Environment="HOME=/home/cloudservice" ExecStartPre=/usr/bin/docker-credential-gcr configure-docker ExecStartPre=/usr/bin/docker run -v /var/log/dataflow/template_launcher:/var/log/dataflow/template_lau ncher -d gcr.io/dataflow-templates-base/template-launcher-logger:stable ExecStart=/usr/bin/docker run --entrypoint /opt/google/dataflow/java_template_launcher -v /var/log/dataflow/template_launcher:/var/log/dataflow/template_lau ncher gcr.io/xxxx/xxxx # ← この後ろにパラメータが追加される runcmd: - systemctl daemon-reload - systemctl start cloudservice.service
35. 解決方法 • パラメータファイルをGCSに置くようにした ◦ launch_flex_template実行前にパラメータファイルを作成して GCSアップロード ◦ GCSのURIだけlaunch_flex_templateのパラメータとして渡す ◦ Pipeline構築前にGCSからファイルを取得してJSON parseする
36. ジョブグラフのサイズにひっかかる
37. The job graph is too large
38. ジョブグラフの制約 • ジョブグラフのサイズが 10 MB を超えてはいけません。パイプライン内の条件によって は、ジョブグラフが制限を超えることがあります。こうした条件のうち一般的なものは次の とおりです。 ◦ 大量のメモリ内データを含む Create 変換。 ◦ リモート ワーカーへの送信のためにシリアル化される大きな DoFn インスタンス。 ◦ シリアル化する大量のデータを（通常は誤って） pull する匿名内部クラス インスタンスとしての DoFn。 これらの状況を回避するには、パイプラインを再構築することを検討してください。 https://cloud.google.com/dataflow/docs/guides/common-errors
39. 解決方法 • パイプラインを分割 ◦ パラメータに応じて出力を分割するオプションがあったので、それぞれ別のジョブで出力するように した
40. その他困ったこと
41. ジョブ起動ユーザと実行ユーザ • Dataflowのジョブ実行時のユーザは2種類ある ◦ Dataflowジョブの作成 ▪ roles/dataflow.admin ◦ Dataflowジョブの実行(ワーカー) ▪ roles/dataflow.worker • カスタムテンプレートやFlexテンプレートを使う場合は権限を分けた方が良い
42. BigQueryIOのエラー処理 • 権限不足などでBigQueryのSQLが実行できなかったときに、ジョブがFailしてくれない • 0件のOutputが出力される
43. Flex Template用REST APIのリファレンスがない • 公式リファレンスにまだ載っていない ◦ https://cloud.google.com/dataflow/docs/reference/rest?hl=ja ◦ githubでgoogle-api-ruby-clientの実装を調べた • サービスアカウントの指定方法が他のAPIと違う ◦ LaunchFlexTemplateRequestのlaunch_parameter.parameters.serviceAccountで指定 ◦ 他の人がGUIから操作していてたまたま発見
44. 今後困りそうなこと • BigQueryIOの結果をBigQueryIOのパラメータに渡すことができない ◦ 1つ目のSQLの結果に応じて2つ目のSQLを構築することができない ◦ 1つ目のSQLだけBigQueryIOで実行し、2つ目はParDoの中で直接BigQueryAPIを叩くと行ける らしい(未確認) ◦ 最悪前段のCollectionの中身分実行されるので呼び出し回数に注意
45. 良かったこと
46. 実行時間の大幅短縮 • 今まで8時間〜24時間ほど(最悪終わらない)かかっていたジョブ(RDS版)が、10分〜50 分で終わるようになった
47. 静的型付け / UnitTest • Ruby + SQLの場合一箇所直すたびにドキドキしながら再実行していた • 入出力が型レベルで保証されるようになったので手戻りが減った ◦ レビューも楽になった • PipelineやFunction単位でUnit Testが書けるようになったため、安心感が大幅に増した • QAでの不具合件数も1桁台で抑えられた
48. まとめ
49. 反省点 • ある程度機能調査はしたつもりだったが、機能要件を網羅できていなかった • 地雷と書いたけど半分くらいは自業自得 • ツール選定、PoCはご安全に