Alexey Romanenko

Alexey Romanenko is Open Source Engineer in Talend (France) with more than 15 years of experience in software development. During his career, he has been working on different projects, like high-load web services, web search engines and cloud storage. Also, he developed and presented a course devoted to Hadoop/Cloud technologies for students. Recently, he joined the Apache Beam project as a new contributor. He spends his spare time with his family and he likes to play ice hockey.

Apache Beamを使用したデータ処理ジョブの開発 – ストリーミングパイプライン

| 2018年8月7日 | Open Source Streaming Data

前回のブログでは、Apache Beamを使ったデータ処理ジョブの開発について紹介しました。今回は、現代のビッグデータ処理で非常にニーズの大きなストリーミングデータの処理について説明します。 バッチとストリーミングの主な違いは、入力データソースのタイプです。データセットが限られていて(サイズが巨大でも)、処理中に更新されない場合は、バッチパイプラインを使用する可能性が高くなります。この場合、入力ソースはファイル、データベーステーブル、オブジェクトストレージ内のオブジェクトなど、何でもかまいません。もう一度強調しますが、バッチ処理では、処理の期間全体でデータが変更可能であり、入力レコードの数は一定です。この点に注意しなければならないのは、ファイルについても、常にファイルの追加や変更が行われるとデータストリームが無限になる可能性があるためです。その場合は、データを処理するためにストリーミングアプローチを適用する必要があります。したがって、データが限られていて不変であることがわかっている場合は、バッチ処理パイプラインを開発する必要があります。 データセットが無制限(継続的に到着)/可変の場合は、処理がより複雑になります。ソースの例としては、メッセージシステム(Apache Kafkaなど)、ディレクトリー内の新しいファイル(Webサーバーログなど)、その他のリアルタイムデータを収集するシステム(IoTセンサーなど)といったものがあります。これらすべての情報源に共通しているのは、常に新しいデータを待たなければならないという点です。もちろん、データを(時間ごとまたはデータサイズごとに)バッチに分割し、分割ごとにバッチ処理することも可能です。しかし、一部の関数については、すべての消費データセットに適用し、そのためのパイプラインを丸ごと作るのが困難です。幸いなことに、この種のデータ処理に簡単に対処できるストリーミングエンジンとして、Apache Spark、Apache Flink、Apache Apex、Google DataFlowを使用できます。これらはすべてApache Beamによってサポートされ、コードを変更することなく、異なるエンジンで同じパイプラインを実行できます。さらに、最小限の変更でバッチ処理でもストリーミングモードでも同じパイプラインを使用できます。入力パイプラインを正しく設定するだけで、即座に使用できます。バッチジョブをストリーミングジョブに書き換えていた頃から、このような機能があれば素晴らしいだろうと考えていました。 理屈はさておき、例を使用して最初のストリーミングコードを記述していきましょう。Kafka(無制限のソース)からデータを読み込み、簡単なデータ処理を実行し、結果をKafkaにも書き戻します。 リアルタイムで到着する地図上のいくつかのオブジェクトの地理座標(XとY)の無限の流れ(この例では、オブジェクトは車だとしましょう)があり、特定地域にあるものだけを選択したい場合を考えます。つまり、我々はKafkaトピックからテキストデータを消費し、それを解析し、指定された制限でフィルタリングし、別のKafkaトピックに書き戻す必要があります。Apache Beamを利用してこれを実現する方法を見ていきましょう。 それぞれのKafkaメッセージには、次の形式のテキストデータが含まれています。 id,x,y このとき: id – オブジェクトの一意のID x, y – 上の座標(整数) 形式に注意し、有効でない場合はスキップします。 パイプラインの作成 前回のブログでのバッチ処理と同じ方法でパイプラインを作成します。 Pipeline pipeline = Pipeline.create(options); Optionsオブジェクトを詳細に指定することで、コマンドラインオプションをパイプラインに渡すことができます。詳しくはGithubを参照してください。 次に、Kafkaの入力トピックからデータを読み込みます。すでに述べたように、Apache BeamはさまざまなIOコネクターを提供しています。KafkaIOもその1つです。したがって、指定されたKafkaトピックからの着信メッセージを消費し、それらをさらに次のステップに伝播する新しい無制限のPTransformを作成します。 pipeline.apply( KafkaIO.<Long, String>read() .withBootstrapServers(options.getBootstrap()) .withTopic(options.getInputTopic()) .withKeyDeserializer(LongDeserializer.class) .withValueDeserializer(StringDeserializer.class)) デフォルトでは、KafkaIOは消費されるすべてのメッセージをKafkaRecordオブジェクトにカプセル化します。ただし、次の変換は新しく作成されるDoFnオブジェクトによってペイロード(文字列値)を取得するだけです。 .apply( ParDo.of( new DoFn<KafkaRecord<Long, String>, String>() …

Read Article