初めに知りたかった Google Cloud Dataflow (Apache Beam) の基礎

はじめに

Google Cloud でバッチ処理を書くときに Google Cloud Dataflow が候補にあがります。Google Cloud Dataflow は、 Apache Beam というフレームワークで書かれたコードを実行するための基盤です。

Apache Beam はデータ処理のためのフレームワークで、Apache Beam のルールに従って書かれたコードは、めちゃくちゃ並列処理できます。

それを Dataflow で動かすことで、大量のサーバーで大量のデータを素早く処理できると言う仕組みです。

そのパワーは凄まじいのですが、代わりに Apache Beam は非常にとっつきづらいものになっています。

そこに拍車をかけるように、Apache Beam のドキュメントは英語しかなく、インターネット上に転がっているナレッジも非常に少ないです。

この記事では、Apache Beam の非常に基礎的な部分だけをまとめます。細かいところまで説明するとややこしいので、そこは Apache Beam の公式ドキュメントを読んでください。この記事を読んだ上で Apache Beam のドキュメントを読むと少しは読みやすくなるでしょう。

参考資料

英語を読むのに抵抗がない人は、公式の Apache Beam Programing Guide を読むのが良いです。

beam.apache.org

Apache Beam の概要

一番基本的な概念は3つです

• Pipeline
• PCollection
• PTransform

PCollection がデータ、PTransform がデータの変換処理です。以下の図のように、PTransform の input に PCollection を与えると、 output として PCollection が返ってきます。

この一連の PTransform の全体を Pipeline といいます。

Pipeline の構築方法

output にどんどん PTransform を apply していくことでパイプラインを構築します。

Pipeline pipeline = Pipeline.create();

PTransform<PBegin, T> transform1 = /* 何らかの PTransform */
PCollection<T> output1 = pipeline.apply(transform1);

PTransform<T, S> transform2 = /* 何らかの PTransform */
PCollection<S> output2 = output1.apply(transform2);

...

ここで PCollection<T> は List<T> と考えてもらうとわかりやすいです。 PTransform<T, S> は、PCillection<T> を PCollection<S> に変換する処理です。

パイプラインは1直線だけでなく、枝分かれしたものも作成することができます。ここは比較的自由にできます。

基本の PTransform である ParDo と DoFn

パイプラインで一番重要なのは「どのような PTransform を適用していくか」です。

よく使う PTransform は Apache Beam 側で用意されています。例えば、 Filter は、条件に合致した要素のみに絞り込む処理です。

しかし、独自の PTrasform を定義したい場合があります。その際に主に使うのが ParDo と DoFn です。この2つはほぼセットで使います。

ParDo は PCollection<T> の中に入っているそれぞれの T に対して処理を適用することを表し、 DoFn は実際にどのような処理を実行するかを定義します。

var output = collection.apply("Print", ParDo.of(new DoFn<String, String>() {
    @ProcessElement
    public void processElement(ProcessContext context) {
        // context.element() で入力された String が取れる
        System.out.println(context.element());
        // context.output すると、出力の PCollection に出力される
        context.output(context.element());
     }
}));

ParDo とか DoFn とか回りくどい書き方になっていますが、こうすることで並列処理とかオートスケールがすごくいい感じに動きます。最初はとっつきにくいですが、分かれば慣れます。

ParDo は PCollection の中にある一つの Element に対して処理を実行するので、ほぼ無限に並列処理できます。

一方で、特定の Element の数をカウントするなど、集計系の処理は ParDo ではできません。集計については、Apache Beam で用意されている Count などの PTransform を使ったり、BatchElements で適当なバッチに分けて処理したり、Tag と言う機能を使ったりする必要があります。

この辺りは難しいので、Apache Beam ここでは説明しません。

Apache Beam であらかじめ用意されている PTransform は、 Transform Catalog で確認することができます。

beam.apache.org

ValueProvider (RuntimeValueProvider / StaticValueProvider / NestedValueProvider)

ValieProvider は、バッチ実行時の引数などを扱うための仕組みです。

RuntimeValueProvider は、パイプライン実行時に決まる値で、PTransform の中で利用する値です。そのため、 RuntimeValueProvider を元にパイプラインの形を変形させることはできません。StaticValueProvider はコンパイル時に決まる値を表しており、StaticValuePriovider に関しては、この値を利用してパイプラインの形を変えることができます。

他の ValueProvider の値を利用して新たな値を計算する場合は NestedValueProvider というものを利用します。例えば、文字列で入力された日付を DateTime 方に変換する場合などですね。

（若干言い方に誤解はありますが）、NestedValueProvider と言う型はなく、元の値が RuntimeValueProvider の場合は、結果の NestedValueProvider も RuntimeValueProvider になります。

https://cloud.google.com/dataflow/docs/guides/templates/creating-templates?hl=ja

SideInput

基本的に、 PTransform は一つの PCollection を入力として受け取ります。PCollection の値以外を受け取りたい場合は SideInput として渡します。SideInput は少しわかりづらい概念ですが、名前の通り補助入力で、SideInput は POollection を渡すと言うよりは、単体の値を渡すイメージです（この辺も誤解がある気がしますが、詳細は Apache Beam のドキュメントを参照してください。機会があれば SideInput については詳細に説明します）。

TupleTag

基本的に PTransform から出力される output は一つの PCollection ですが、複数の複雑な output を渡したい場合は TupleTag と言うものを使います。

output に対してタグをつけて、タグで複数の ouput を区別するイメージです。

Composite Transform

PTransform を入れ子にすることができます。詳細は以下のドキュメントを見てください。Google Cloud Dataflow 上のグラフが見やすくなったり、抽象化によりコードの可読性を上げることができます。

beam.apache.org

Create.of / Create.empty

動作確認やテストで、Datastore 等からデータを取得してくる代わりに、固定のデータを使いたい場合に使えます。

// 固定の PCollection を出力したい場合
Create.of("A", "B", "C");

// 空の PCollection を出力したい場合
// empty だと型がわからないので、 TypeDescriptor で型を渡す必要がある
Create.empty(TypeDescriptor.of(String.class));

TestPipeline

テストを描きたいときに使えます

まとめ

とりあえず以下を理解する

• PTransform
• PCollection
• ParDo / DoFn

集計とか結合とか、その辺の複雑なやつは、余裕があったら説明します

まずはドキュメント読むのがおすすめです

beam.apache.org