PLAID Advent Calendar 2018 の17日目
今回はCloud Dataflow SDKからApache Beam SDKへの移行についてです。
13 日目のビジネスサイドが KARTE Datahub を用いて学ぶ、初めての SQLで紹介されているKARTE Datahubのバックエンドの一部はDataflowで動かしています。
Dataflowジョブ実行部分の開発は昨年末頃から始めて、今年の4月頃からはプロダクション環境でも動かし始めました。
プロダクション環境での今月のバッチジョブ実行数は多い日で500回を超えています 
注意
今回はJava SDKの話しかしません。
Pythonの話を期待された方は申し訳ございません 
Cloud Dataflow SDKについて
Cloud Dataflow SDKはGoogleがCloud DataflowでApache Beamのパイプラインを実行できるようにするために提供しているものです。
Apache Beamのパイプライン実行環境はDataflowの他にもいくつかありますが、Dataflowで動かすのに必要な機能をパッケージ化したもの(Apache Beam SDKのサブセット)ということになります。
パイプライン実行環境についてはこちら
Cloud Dataflow SDKのサポート終了のお知らせ
Apache Beam と Cloud Dataflow のヘルプ
Cloud Dataflow SDK のサポート終了のお知らせ: Cloud Dataflow SDK 2.5.0 が Cloud Dataflow SDK の最後のリリースになります。Cloud Dataflow SDK のリリースは Apache Beam SDK のリリースとは別個のものです。Cloud Dataflow サービスでは、正式な Apache Beam SDK のリリースがフルサポートの対象となります。Cloud Dataflow サービスでは、バージョン 2.0.0 以降のリリース済みの Apache Beam SDK のリリースもサポートされます。
また、Eclipseプラグインの動作も2.5.0までとなっています。
Java と Eclipse を使用したクイックスタート
Cloud Dataflow Eclipse プラグインは、Cloud Dataflow SDK ディストリビューションのバージョン 2.0.0～2.5.0 でのみ動作します。Cloud Dataflow Eclipse プラグインは、Apache Beam SDK ディストリビューションでは動作しません。
Eclipseからパイプラインの実行ができるのはとても便利だったので切ないですね
Apache Beam SDKへの移行
上の方でCloud Dataflow SDKはApache Beam SDKのサブセットであると書きました。
Cloud Dataflow SDKがパッケージしているものを調べればApache Beam SDKに移行できますね。
Maven Central Repositoryで検索。ありました。
com.google.cloud.dataflow:google-cloud-dataflow-java-sdk-all
dependenciesを抜粋
<dependencies>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-core</artifactId>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-direct-java</artifactId>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-kafka</artifactId>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
なるほど 
google-cloud-dataflow-java-sdk-allを置き換えてみます。
変更前
<dependencies>
<dependency>
<groupId>com.google.cloud.dataflow</groupId>
<artifactId>google-cloud-dataflow-java-sdk-all</artifactId>
<version>2.5.0</version>
</dependency>
</dependencies>
変更後
<dependencies>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-core</artifactId>
<version>2.5.0</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
<version>2.5.0</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-direct-java</artifactId>
<version>2.5.0</version>
</dependency>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-runners-google-cloud-dataflow-java</artifactId>
<version>2.5.0</version>
</dependency>
</dependencies>
Maven buildで吐き出されたjarを叩いて、無事にパイプラインの実行ができました 
余談ですがversionを変数にしておくとSDKのversionを変更するときに便利です。
<properties>
<beam.version>2.8.0</beam.version>
</properties>
<!-- propertiesに定義すると↓のようにかけます -->
<dependencies>
<dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-core</artifactId>
<version>${beam.version}</version>
</dependency>
</dependencies>
おまけ
Apache Beam SDK 2.9.0
今月13日(4日前ですね)にApache Beam SDK 2.9.0がリリースされています。
https://beam.apache.org/get-started/downloads/#releases
2.4.0から2.5.0の間が3ヶ月くらい開いたりしてたので、ここ数ヶ月のリリースの早さに勢いを感じます。
開発小話
Cloud Spannerへのマルチバイト文字の書き込みができないバグを踏んだり(Apache Beam SDK 2.7.0で解消されています)
Dataflow WokerのAutoscalingが優秀でthroughputが出すぎてCloud Bigtableへの書き込み時にHotspotが発生しがちだったり
Dataflow Workerが1から1000にスケールする謎の挙動が発生してCompute Engineの制限に引っかかりジョブが失敗したり
他にも色々な問題にぶち当たりましたが現在は安定して稼働しています。
問題への対処は学びが多くて、大変でしたが楽しかったですね。
おわりに
社内ではDatahubでDataflowを使い始めてから他のプロジェクトでも使われるようになり活用の幅が広がってきました。
Datahubではバッチジョブしか実行していませんがストリームジョブを動かしているところもありますし、今後もさらに活用の機会、幅が広がっていきそうです。
今年に入ってGo SDKがサポートされたりと活発な印象を受けるApache Beamコミュニティの動向に今後も注目していきたいです。