Beamを使ってPipelineを構築します。次回、自分の得意なGolangで書くエントリを執筆します。Stormなどのストリーム処理システムでは一般的なJavaを使おうと思います。Quickstartでは、なぜかPythonやGolangはSDKをインストールしたり環境構築から始まるのに対して、Javaは文字数カウントサンプルをやらさせます。Mavenがインストールされていることを前提とします。-DinteractiveMode=falseそしてディレクトリに移動します。word-count-beamそして、以下のようなディレクトリ構造になっています。データのソースsrc内のコードを見てみると、どうやらシェイクスピアの文献の文字数を数えているようです。Storageにあるようです。Beamが元はGoogleのものだったので納得ですね。Storageの公開URLは以下のようになります。https://storage.googleapis.com/{プロジェクト名}/{バケット名}/{...バケット内の構造(パス)}プロジェクト名やバケット名は以下で指定できるみたいです。コメントがありました。dataflowOptions.setTempLocation("gs://SET_YOUR_BUCKET_NAME_HERE/AND_TEMP_DIRECTORY");そして読み込みます。p.apply(TextIO.read().from("gs://apache-beam-samples/shakespeare/*"))これによって、シェイクスピアのファイルを一行ずつ読み込みます。戻り値にはPCollectionを返します。Trasform以下のようなTransformがPipelineとして定義されてあります。.apply(TextIO.write().to("wordcounts"));何をしているかというと、まず、FlatMapElements変換をして、PCollection<String>に変換されます。Filterによって空文字を除去している。次にCountで文字数をカウントしています。最後にMapElementsで出力するためにMap要素にしています。ローカルで実行してみる以下のコマンドを実行して、生Beamとして実行します。-Pdirect-runnerファイルがいくつも生成されているので、確認します。Beamのプログラミングモデルまず、はじめにBeamの重要なクラスを解説します。BeamのPipelineそのものである。分散データセットを表現する。bounded(有限)でもunbounded(ストリーム)の状態でも良い。PCollectionを入力に、データのtransformを行うものである。Pipelineオブジェクトを作成する以下のようにoptionsを作成して、Pipelineを作成します。PCollectionを作成するインメモリでも、テキストでもどちらでも可能です。Transformを作成するBeamでは、いくつもの種類のTransformを作成できます。以下のような構文で作ることができる。PCollection].apply([Transform])これを繰り替えることによって様々なTransformを定義できるのである。同じPCollectionに対して処理をかけたいときは以下のように書きます。2])また、コアtransformがサポートされている。一般的な並列処理である。HadoopのMapに当たる。key/valueのコレクションの処理をする。一般的なflattenをする。小さなコレクションに分割する。Mapのようなものである。applyを@Overrideをしなければならず、MapElements.via(simpleFunction)をして一つずつ実行される。ParDoで実行される。ストリームを読み込んだりするのに使われる。Beamで重要なことは、DoFnをたくさんextendして定義していくことになることになる。ParDoまず、クラスを定義する。ここで<String>は入力で、Integerは出力である。Elementで入力を定義する。出力はOutputReceiverで渡す。実際にapplyしてみる。ComputeWordLengthFn()));戻り値にはPCollectionが返ってくる。GroupByKey以下のようなコレクションがあるとする。6...そのときに出力は以下のようになる。同じユーザー名などのキーで、いくつかvalueを持つようなものである。結果として以下のようなものである。Combineコレクションを結合させるために使われる。簡単な関数は以下のように定義できる。Flatten同じデータタイプを持つような複数のPCollectionを一つのPCollectionにマージする。概念的には、以下のようなものである。PartitionこれはFlattenと正反対の処理になる。概念的な処理は以下のようになる。一般的には、以下の制限がある。データはシリアライズ可能である。データは並列処理可能である。Trasfrom今までは、プレーンテキストを読み込んでいました。Avroでシリアライズされていたり、いろんなソースが想定できます。Javaでは以下のようなソースに対応しています。Beamは流行っているし、かなりドキュメントも質がいいPipelineとしてのアプリケーションクラスにTransformもexec関数も実装すると責任を明確にできるGoogle発ってことで安心できる。ここの節は余談なので飛ばしてもらっていいです。Kubernetesのいろは。»プロフィール学生エンジニア。忘れてしまうので備忘録です。flaskでServerlessにホストするTelepresence概論スターをつけました引用をストックしましたストック一覧を見る閉じる引用するにはまずログインしてくださいログイン閉じる引用をストックできませんでした。再度お試しください閉じる限定公開記事のため引用できません。