Beamに関するシリーズの第1部です。知らなくとも、決して恥ずべきことではありません。Beamは、まだデータ処理の世界では比較的なじみの薄いものです。Beamについて学ぶことが楽しくなり、その素晴らしさがわかってきました。Beamは、バッチとストリーミングのデータ処理ジョブを実装し、さまざまなIOのセットを使用して任意の実行エンジンで実行するための簡単な方法を提供する統一プログラミングモデルです。Beamに一連のブログ記事を掲載することにしたのはこのためです。Beamを使用したデータ処理ジョブのユースケースを紹介します。今回のトピックはバッチ処理です。次の例を見てみましょう。自動車販売店での一定期間内の自動車販売を分析したいとします(たとえば、各ブランドの自動車の販売台数など)。これは、データセットが制限され(有限量のデータ)、更新されない(過去に売上が発生した)ことを意味します。この場合のデータ分析にはバッチプロセスを使用できます。入力データとして、以下の形式の販売車のテキストログがあります。id,brand_name,model_name,sales_number例:1,Toyota,Prius,32,Nissan,Sentra,23,Ford,Fusion,4初めてBeamアプリケーションの実装を開始する前に、常に使用される中心的概念について認識しておく必要があります。Beamには、パイプライン、PCollection、PTransformの3つの主な概念があります。パイプラインは、データ処理タスク全体のワークフローを最初から最後までカプセル化します。PCollectionは、BeamがPTransform間でデータを転送するために使用する分散データセットの抽象化です。PTransformは、入力データ(入力PCollection)を操作して出力データ(出力PCollection)を生成するプロセスです。通常、最初と最後のPTransformは、制限されている(バッチ処理)または制限されていない(ストリーミング処理)データを入出力する方法を表します。簡単にするために、パイプラインはワークフロー全体を表すDAG(有向非巡回グラフ)として、PTransformはノード(データを変換する)として、PCollectionsはこのグラフのエッジとして考えることができます。Guide」を参照してください。例に戻って、提供されたデータセットを処理する最初のパイプラインを実装してみましょう。パイプラインの作成まず、新しいパイプラインを作成します。Pipeline.create();次に、pipeline.apply()メソッドを使って新しいPTransformを作成しましょう。これはテキストファイルからデータを読み込み、文字列の新しいPCollectionを作成します。これを行うには、Beamですでに実装されているIOの1つであるTextIOを使用します。TextIOはテキストファイルの1行ずつの読み書きを可能にします。さまざまなファイルシステムの使用、ファイルパターンのサポート、ファイルのストリーミングなど、ほかにも多くの機能があります。Beamのドキュメントを参照してください。apply(TextIO.read().from(“/path/to/input/file”))このPTransformの出力は、PCollectionの新しいインスタンスであり、集合の各エントリは入力ファイルのテキスト行です。結果としてブランド別の総販売数を求めたいので、それらをグループ化する必要があります。したがって、次のステップでは、すべての行を解析し、keyがブランド名でvalueが販売数となるキー/値ペアを作成します。前のPTransformからの出力PCollectionが、ここでは入力PCollectionになります。このステップでは、Beamの内部PTransform(MapElements)を使用して、提供されるSimpleFunctionインターフェイスの実装を使用して、入力項目ごとに新しいキー/値ペアを作成します。次に、別のBeamの変換であるGroupByKeyを使用してブランド別に販売数をグループ化します。出力結果となるキー/値のPCollectionは、keyがブランド名で、valueがそのブランドの反復可能な売上高の集まりです。Integer>create())これで、ParDo変換の独自の実装を使用して、ブランド別の自動車販売数を集計する準備が整いました。パイプラインを完成させるために、別のIO変換を適用し、文字列のPCollectionを取得してテキストファイルに書き込みます。.apply(TextIO.write().to(“/path/to/output/dir”).withoutSharding());最後に、作成したパイプラインを実行します。pipeline.run();とても簡単ですね。Hadoopに精通している方は、このパイプラインが以下の点で何かに似ていることにお気づきかと思います。このシンプルなパイプラインは、従来のMapReduceジョブで実行できます。Beamの方が大幅に単純で明確です(Javaであるにもかかわらず!)。別の変換を追加してパイプラインを拡張する場合も、それほど複雑化することはありません。前述したように、Beamパイプラインはさまざまなランナー(処理エンジン)で実行できます。Dataflowこれを行うには、MavenまたはGradleプロジェクト構成に対応する依存関係を追加するだけです。これには、各ランナーで実行するために、パイプラインコードを調整したり書き直したりする必要がないというメリットがあります。以前に必要なランナーの依存関係がすべて含まれていた場合は、jarを再コンパイルする必要はありません。使用するランナーを選択するだけでいいのです。Runnerは、通常はパイプラインをテストするために使用されるローカルランナーです。Runnerへの依存関係を指定する必要があります。<dependency><groupId>org.apache.beam</groupId><artifactId>beam-runners-direct-java</artifactId><version>2.3.0</version><scope>runtime</scope></dependency>その後、プロジェクトをコンパイルする必要があります。packageそして、ダイレクトランナーでパイプラインを実行します。-Dexec.args="--runner=DirectRunner”たとえば、入力ファイルに次のデータが含まれているとします。/tmp/beam/cars_sales_log1,Toyota,Prius,32,Nissan,Sentra,21,Toyota,Yaris,43,Ford,Fusion,53,Ford,Kuga,3すると、最終結果は次のようになります。8サポートされているすべてのランナーのリストと指示、使用方法は、このページをご覧ください。最後に、この例で使用したすべてのコードは、すべてGitHubレポジトリ(https://github.com/aromanenko-dev/beam-tutorial)に公開されています。このブログ記事シリーズの次回は、Beamでのストリーミングデータ処理について説明します。無制限のデータソースを使用したデータ分析タスクの例を取り上げ、Beamの機能を確認します。利用されているリソースどこから始めてよいかわからないという方へ。