Apache Flinkによるストリーム処理

第5章. DataStream API (v1.7)

この作品はAIを使って翻訳されている。ご意見、ご感想をお待ちしている：[e-mail]

この 章では、Flink の DataStream API の基本を紹介する。典型的な Flink ストリーミングアプリケーションの構造とコンポーネントを示し、Flink の型構造とサポートされるデータ型について説明し、データの変換とパーティション分割について述べる。Windows演算子、時間ベースの変換、ステートフル演算子、コネクタについては次の章で説明する。この章を読めば、基本的な機能を備えたストリーム処理アプリケーションの実装方法がわかるだろう。コード例では簡潔にするためにScalaを使用しているが、Java APIもほぼ同様である（例外や特殊化については指摘する）。また、JavaとScalaで実装された完全なサンプル・アプリケーションをGitHubリポジトリで提供している。

こんにちは、フリンク！

、DataStream APIを使ってストリーミング・アプリケーションを書くことがどのようなものかを第一印象で理解するために、簡単な例から始めてみよう。この例を使って、Flink プログラムの基本構造を紹介し、DataStream API の重要な機能を紹介する。このサンプル・アプリケーションは、複数のセンサーから温度測定値のストリームを取り込む。

まず、センサーの読み取りを表すデータ型を見てみよう：

case class SensorReading(
  id: String,
  timestamp: Long,
  temperature: Double)

例5-1のプログラミングは、温度を華氏から摂氏に変換し、各センサーの平均温度を5秒ごとに計算する。

例5-1. センサーのストリームについて5秒ごとの平均温度を計算する

// Scala object that defines the DataStream program in the main() method.
object AverageSensorReadings {

 // main() defines and executes the DataStream program
 def main(args: Array[String]) {

   // set up the streaming execution environment
   val env = StreamExecutionEnvironment.getExecutionEnvironment

   // use event time for the application
   env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

   // create a DataStream[SensorReading] from a stream source
   val sensorData: DataStream[SensorReading] = env
     // ingest sensor readings with a SensorSource SourceFunction
     .addSource(new SensorSource)
     // assign timestamps and watermarks (required for event time)
     .assignTimestampsAndWatermarks ...