Apache Flink と Kafka の統合

Apache Flink は、有限および無限のデータストリームに対する状態を持つ計算のための分散処理エンジンです。イベントを到着時に消費し、アプリケーション状態を維持し、イベント時刻を使用し、障害から復旧し、結果を継続的に公開できます。

Apache Kafka は、分散イベントストリーミングプラットフォームです。プロデューサーはパーティション分割されたトピックにレコードを追加し、Kafka はそれらのレコードを保持し、コンシューマーは自身のペースで読み取ります。ストリーミングアーキテクチャでは、Kafka は永続的なトランスポート層として機能し、Flink は計算を実行するのが一般的です。

applications -> Kafka topics -> Flink job -> Kafka topics / databases / APIs

2 つのシステムは異なる問題を解決します。Kafka はイベントログを保存および転送します。Flink は、それらのログをフィルタリング、エンリッチメント、集計、結合、パターン検出によって結果へ変換します。プロセス、マシン、ネットワークの障害後も計算が状態を保持し、正確性を維持する必要がある場合、この統合は特に有用です。

用語集 #

ストリームの種類 #

ストリームの種類は、主に source の設定を通じて Job コードに影響します。下流の DataStream コードは、多くの場合そのまま利用できます。

// Unbounded: continuously discovers and consumes Kafka records
KafkaSource<Event> liveSource = KafkaSource.<Event>builder()
        .setTopics("events")
        .setStartingOffsets(OffsetsInitializer.committedOffsets())
        .build();

// Bounded: finishes at specified partition offsets
KafkaSource<Event> replaySource = KafkaSource.<Event>builder()
        .setTopics("events")
        .setStartingOffsets(OffsetsInitializer.offsets(startOffsets))
        .setBounded(OffsetsInitializer.offsets(endOffsets))
        .build();

どちらも同じパイプラインに入力できます。

env.fromSource(source, watermarks, "events")
   .keyBy(Event::customerId)
   .process(new CustomerProcessor())
   .sinkTo(sink);

boundedness は、どのようなコードが意味を持つかにも影響します。

• Unbounded stream では、count や aggregate などの操作に window または別の出力ポリシーが必要です。
• グローバルな「すべてをソートする」処理は、unbounded stream では完了できません。
• Bounded job では、入力の終端後に最終結果を生成できます。
• 継続的なファイル監視によって、通常は bounded な file source が unbounded になります。

ランタイムの動作も選択できます。

env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
env.setRuntimeMode(RuntimeExecutionMode.BATCH);
env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

Source の boundedness はジョブを終了できるかどうかを決定し、runtime mode は Flink がデータをスケジュールおよび交換する方法を決定します。

Flink と Kafka を使う理由 #

• Flink は Kafka なしでも動作します。Unbounded stream は、ファイルまたはファイル群を意味することもあります。
• Kafka は Flink なしでも動作します。従来のカスタム Kafka コンシューマーをデプロイできます。

基本的な Kafka コンシューマーはレコードを読み取り、変換できます。しかし、実際のストリーミングアプリケーションでは、次の機能が必要になることがあります。

• 1 台のマシンのメモリを超える可能性がある keyed state
• 遅延イベントや順不同イベントに対応する event-time window
• 入力 offset、operator state、出力をまたぐ一貫した復旧
• state を手動で再分配せずに行う再スケーリング
• ストリーム間、またはストリームと参照データ間の結合
• 通常の処理を停止しない非同期チェックポイント
• 継続的に到着するレコードに対する SQL
• 運用メトリクス、savepoint、制御されたアップグレード

Flink は、これらの機能を 1 つの実行モデルの一部として提供します。Kafka は再生可能な入力と永続的な出力を提供するため、障害が発生した Flink ジョブは最初からやり直したり、どのレコードが処理済みかを推測したりするのではなく、整合性のある地点から再開できます。

Flink がカスタム Kafka コンシューマーより自動的に優れているわけではありません。各レコードが独立しており、処理が次のようなものだけである場合、単純なコンシューマーの方が適しています。

consume -> perform one stateless operation -> publish or store

それ以外の場合、カスタムサービスは状態の永続化、offset の調整、backpressure、再試行、event-time 処理、再スケーリング、復旧を実装または統合する必要があります。これらの懸念が複数計算の一部になると、Flink の価値が高まります。その代わり、通常のサービスをデプロイする場合と比べ、Flink クラスターを実行または利用する運用コストが発生します。

適合するメンタルモデルとユースケース #

Flink ジョブは、イベントストリームに対して継続的に評価される分散データベースクエリとして考えることができます。

unbounded input -> transformations and durable state -> continuous output

Kafka はイベント履歴を保持および転送します。Flink は、その履歴から派生状態と結果を継続的に計算します。このモデルは、次の用途に適しています。

• 不正検知: 離れた場所で短時間に発生したカード取引の急増を検知する
• リアルタイム分析: 売上、アクティブユーザー、コンバージョンのメトリクスを継続的に更新する
• ストリーミング ETL: 注文を検証し、顧客または商品データでエンリッチし、分析用レコードを公開する
• レコメンデーション: 検索、クリック、購入から買い物客の関心を更新する
• 監視: サービスイベントを相関付け、移動失敗率がしきい値を超えた場合に通知する
• 重複排除: イベント識別子を記憶し、再試行された支払いや通知を二重処理しない
• セッション化: 非アクティブ期間によってセッションが閉じるまで Web アクティビティをグループ化する
• ストリーム結合: 遅延または順不同イベントを含め、広告クリックと後続の購入を関連付ける
• 物流: GPS 更新と経路データを組み合わせて到着予定時刻を更新する
• IoT 処理: センサー値を集計し、異常パターンを検知する

Kafka なしの Flink とストリーミング優先設計 #

Flink は Kafka を必要としません。Kafka は、Pulsar、Kinesis、データベースの change-data-capture フィード、ファイル、Iceberg テーブル、アプリケーション固有のコネクターなどの source および sink の候補の 1 つです。

database CDC -> Flink -> Iceberg
Kinesis      -> Flink -> database
HDFS files   -> Flink -> HDFS files
IoT events   -> Flink -> alerts

Kafka がない場合でも、Flink は Apache Spark と大きく重なります。特に bounded SQL や ETL ワークロードではその傾向があります。主な違いは絶対的な機能境界ではなく、中心となる設計思想です。

有限のファイル変換やデータレイク分析は、Spark に自然なワークロードであることが多いです。Kafka がなくても、アプリケーションが継続的な state、細粒度の時間セマンティクス、低レイテンシ、または長寿命のイベント駆動動作を必要とする場合、Flink は独自の位置付けを持ちます。

ストリーミング機能と Flink による実装 #

Flink は、すべての source コンシューマーが個別に実装する必要がある機能としてではなく、state、time、recovery、flow control をランタイムの関心事として扱います。

管理対象 state は論理的にはオペレーターに、keyed state の場合は現在のアプリケーション key に紐付けられます。ランタイム state は JVM heap または組み込み state backend に存在し、checkpoint は整合性のあるスナップショットを永続ストレージへコピーします。復旧または再スケーリング時に、Flink はアプリケーションが手動で再構築するのではなく、state を復元して再分配します。

Event-time timer は壁時計の時刻が進んだときではなく、watermark がそのタイムスタンプを通過したときに発火します。Window は allowed lateness の期間中に結果を保持し、遅延イベントが到着したときに更新できます。そのポリシーより遅いレコードは side output にルーティングできます。Processing-time timer は TaskManager の時計に従い、壁時計の動作が実際の要件である場合に有用です。

Flink の基本概念 #

システムを接続する前に、Flink のモデルを理解しておくと役立ちます。

Job と Operator #

Flink アプリケーションはオペレーターのグラフを定義します。

source -> transform -> partition -> window / process -> sink

Flink はこのグラフを分散ジョブに変換します。各オペレーターは複数の並列サブタスクを持てます。レコードは入力データセット全体が完了するまで待つのではなく、グラフを継続的に流れます。

DataStream コードでは、operator は論理的な処理ステップであり、transformation はそのステップをジョブグラフに追加する API 操作です。両者はしばしば同じ意味で使われます。Flink は効率化のため互換性のあるオペレーターを 1 つのランタイムタスクにチェーンできるため、オペレーターがスレッド、プロセス、ネットワーク交換と 1 対 1 に対応するとは限りません。

Source と Sink Operator #

Source はジョブにレコードを導入します。また、シリアライゼーション、パーティション検出、boundedness、復旧を継続する位置を定義します。

本番コネクターでは通常 fromSource を使用します。Collection とメモリ内 source は、主に例とテストで役立ちます。

Sink は Flink の外部へ結果を送信します。

Source と sink はジョブの外部境界を形成します。Sink の配送保証は、コネクターと checkpoint 設定の両方に依存します。map、process、その他の transformation 内から外部システムへ書き込むと、sink プロトコルを迂回するため、checkpoint と自動的に調整されません。

基本的な Transformation Operator #

基本的な transformation はレコードを独立して処理し、通常は管理対象 state を必要としません。

メモリ内ですでに利用可能な解析やエンリッチメントには map、複数の結果を生成する可能性があるトークン化や検証には flatMap、不要なレコードの破棄には filter を使用します。returns は transformation の型を記述するものであり、レコード自体を処理するものではありません。

Partitioning Operator #

Partitioning は各レコードを受け取る下流サブタスクを決定します。ネットワーク交換が発生する場合があります。

keyBy は最も重要な partitioning 操作です。KeyedStream を作成し、keyed state、timer、keyed window、keyed aggregation を有効にします。Kafka の partitioning とは異なります。Kafka は Flink が読み取る前にレコードを source partition へ割り当てますが、keyBy は Flink ジョブ内部でレコードを再パーティション化します。

Aggregation Operator #

Aggregation は複数のレコードを継続的に更新される結果へ統合します。一般的な API は次のとおりです。

KeyedStream 上の window なし aggregation は無期限に更新されるため、key ごとに 1 つの aggregate を保持します。Windowed reduce と aggregate は、aggregation を各 window に限定します。accumulator を入力と結果の両方と異なる型にする必要がある場合、たとえば合計と件数を保持して平均を生成する場合は、aggregate を優先します。

Window Operator #

Window は unbounded stream を計算用の有限バケットにグループ化します。頻繁に使用される window assigner は次のとおりです。

keyBy の後では window(...) を使用し、key がない場合は windowAll(...) を使用します。頻繁に使われる window 計算は reduce、aggregate、process です。Incremental な reduce または aggregate は、window 内のすべてのレコードへのアクセスが必要になる可能性のある process より一般に効率的です。Flink がレコードを段階的に集計し、ProcessWindowFunction が最終結果に window メタデータを追加する形で組み合わせることもできます。

Process Operator #

Process function は、管理対象 state、timer、timestamp、side output を必要とするロジックに対する Flink の低レベルな拡張点です。

一般的に使用される API 呼び出しは process(...) です。Side output は OutputTag で宣言し、function context を通じて出力します。これは遅延イベント、無効レコード、個別にルーティングするアラートに有用です。state と timer が不要なら、より単純な transformation を使用します。

複数ストリームと Join Operator #

これらのオペレーターは関連するストリームを結合します。

union はレコードをインターリーブしますが、結合は行いません。connect は 2 つの入力にまたがる共有ロジックまたは state を可能にしますが、レコードを自動的に相関付けることもありません。レコードを key と時刻で照合する必要がある場合は、intervalJoin、windowed join、または KeyedCoProcessFunction を使用します。

非同期 I/O Operator #

非同期 I/O によって、タスクスレッドをブロックせずに多くの外部リクエストを同時に実行できます。

どちらも AsyncFunction または RichAsyncFunction に加え、timeout と capacity を使用します。ストリームまたは broadcast state として表現できないリモート lookup によく使われます。外部クライアント自体が非同期でなければなりません。ブロッキング呼び出しをラップしても nonblocking にはなりません。timeout、retry、ordering、replay の動作を明示的に定義します。

DataStream と特殊化された型 #

org.apache.flink.streaming.api.datastream.DataStream<T> は、レコードが型 T を持つストリームの基本 Java 抽象です。すべてのレコードをメモリ内コレクションとして保持するものではなく、ジョブグラフ内の論理 transformation へのハンドルです。map、keyBy、sinkTo などのメソッドを呼び出すと、そのグラフへ transformation が追加され、別の API オブジェクトが返されます。

Flink は、ストリームの生成方法またはパーティション方法によって次に有効な操作が異なる場合、特殊化された DataStream サブタイプを返します。

階層は次のように要約できます。

DataStream<T>
├── SingleOutputStreamOperator<T>
│   └── DataStreamSource<T>
├── KeyedStream<T, K>
├── SideOutputDataStream<T>
└── CachedDataStream<T>

宣言された Java 型によって、利用可能な fluent method が決まります。たとえば次のようになります。

SingleOutputStreamOperator<Purchase> purchases = raw
        .map(OrderPipeline::parsePurchase)
        .name("parse-purchase");

KeyedStream<Purchase, String> byCustomer =
        purchases.keyBy(Purchase::customerId);

SingleOutputStreamOperator<Purchase> totals = byCustomer
        .reduce(OrderPipeline::addAmounts)
        .uid("customer-totals");

name と uid は SingleOutputStreamOperator で利用でき、keyed reduce は KeyedStream で利用できます。keyBy の結果を通常の DataStream<Purchase> 変数へ代入すると、コンパイル時に KeyedStream 固有のメソッドが隠れます。後からキャストするのではなく、それらのメソッドが必要な間は特殊化された型を保持します。

DataStream と関連する重要な fluent API 型がいくつかありますが、そのサブクラスではありません。

これらの wrapper および builder 型によって、無効な操作順序を表現しにくくなります。たとえば WindowedStream は一般的な DataStream 操作すべてではなく window 計算を公開し、DataStreamSink は出力ブランチの終端を示します。

Runtime Component #

Flink はジョブの調整と実行を分離します。

• Dispatcher は送信されたジョブを受け付け、それぞれのジョブに JobMaster を起動します。
• JobMaster は 1 つのジョブを調整します。実行グラフを構築し、slot を要求し、サブタスクをスケジュールし、checkpoint と復旧を調整します。
• Flink ResourceManager は基盤プラットフォームからワーカーリソースを取得および追跡します。
• TaskManager は Flink のワーカープロセスです。オペレーターサブタスクを実行し、他の TaskManager とレコードを交換します。
• task slot は TaskManager の実行容量単位です。通常、1 つの TaskManager には複数の slot があります。

JobManager は、Flink のコントロールプレーンプロセスをまとめて指す名前として一般的に使用されます。デプロイモードによっては、そのプロセスに Dispatcher、Flink ResourceManager、REST endpoint、1 つ以上の JobMaster が含まれます。Kubernetes TaskManager pod または YARN TaskManager container はプラットフォームのリソース境界であり、TaskManager はその内部の Flink ワーカープロセスです。

Bounded と Unbounded Stream #

Unbounded stream には事前定義された終端がありません。ライブイベントに使用される Kafka topic は通常 unbounded であるため、Flink ジョブは実行を継続し、新しいレコードを処理します。

Bounded stream には有限の終端があります。Flink は Kafka offset の固定範囲を読み取り、batch として処理することもできます。したがって、同じ API で継続的なパイプラインと有限の再処理ジョブの両方をサポートできます。

Keyed State #

keyBy は customerId などの論理 key によってレコードをパーティション化します。同じ key のすべてのレコードは同じ下流サブタスクに送られます。

events
    .keyBy(Purchase::customerId)
    .process(new CustomerPurchaseFunction());

Process function は各顧客について独立して state を維持できます。Flink は異なる parallelism でジョブを復元するときに、その state を再分配します。

Event Time と Watermark #

Processing time は Flink がレコードを処理する時刻です。Event time はイベントが発生した時刻で、通常はレコード自身に含まれます。

Kafka レコードが遅延したり順不同で到着したりする場合、event time によってより安定した結果が得られます。Watermark は event time がどこまで進んだと Flink が判断しているかを示します。Bounded out-of-orderness 戦略によって、一定の遅延を許容できます。

WatermarkStrategy<Purchase> watermarks =
        WatermarkStrategy
            .<Purchase>forBoundedOutOfOrderness(Duration.ofSeconds(10))
            .withTimestampAssigner(
                (purchase, previousTimestamp) -> purchase.occurredAt()
            );

Watermark が window の終端を通過すると、Flink はその window の結果を出力できます。Watermark の後に到着したイベントには、allowed-lateness policy または side output が必要になる場合があります。

Checkpoint と Savepoint #

Checkpoint は次の整合性のあるスナップショットです。

• Kafka source position
• operator と keyed state
• timer
• サポート対象 sink の transaction state

ジョブが失敗すると、Flink は最後に完了した checkpoint を復元し、そこに記録された Kafka offset から再開します。

Savepoint は、アップグレード、移行、計画的な再スケーリングなどの運用操作を目的とする、ユーザーが起動するスナップショットです。Checkpoint と savepoint は類似した state の仕組みを使用しますが、ライフサイクルと目的が異なります。

デプロイメントとコンポーネントのライフサイクル #

Flink は YARN に依存しません。Kubernetes、YARN、スタンドアロンクラスター、ローカル JVM プロセス、マネージド Flink サービスで実行できます。Resource manager は Flink プロセスを起動し、Flink 自体がオペレーターをスケジュールし、state を管理し、checkpoint を調整します。

Flink プロセスに本質的に永続的なものはありません。ストリーミングジョブは無期限に実行される場合がありますが、JobManager、TaskManager、サブタスクの実行試行は交換可能です。有用な区別は、復旧後も state が存続できる論理ジョブと、現在それを実行しているプロセスの実体です。

Flink cluster
|
+-- control-plane process
|   +-- Dispatcher                 cluster lifetime
|   +-- Flink ResourceManager      cluster lifetime
|   +-- REST endpoint / blob store cluster lifetime
|   `-- JobMaster                  job lifetime
|       +-- scheduler              job lifetime
|       +-- checkpoint coordinator job lifetime
|       `-- operator coordinators  job lifetime
|
`-- TaskManagers                   dynamically allocated worker lifetime
    `-- task slots                 TaskManager lifetime
        `-- operator subtasks      execution-attempt lifetime

デプロイモードによって、外側のクラスターライフサイクルが決まります。

• session mode では、1 つの長寿命クラスターが複数のジョブを受け付けます。Dispatcher と Flink ResourceManager は session が停止するまで存続し、各ジョブは自身のライフサイクルだけ JobMaster を持ちます。需要に応じて TaskManager を取得および解放できます。
• application mode では、クラスターは 1 つのアプリケーションにスコープされ、1 つ以上のジョブを送信できます。ユーザーの main メソッドはクラスター内で実行され、アプリケーションが完了すると通常クラスターも終了します。
• 旧式の per-job mode では、送信された 1 つのジョブに対して 1 つのクラスターが作成され、ジョブとともに終了します。現在の Flink デプロイメントでは一般に application mode が推奨されます。

┌─────────────┬────────────────────────────────────────────────────────────────────────┐ │ Mode │ JobManager lifetime │ ├─────────────┼────────────────────────────────────────────────────────────────────────┤ │ Session │ Cluster lifetime; survives jobs finishing or being canceled │ ├─────────────┼────────────────────────────────────────────────────────────────────────┤ │ Application │ Application lifetime; usually ends when all application jobs terminate │ ├─────────────┼────────────────────────────────────────────────────────────────────────┤ │ Per-job (LEGACY DON’T USE) │ Single job lifetime │ └─────────────┴────────────────────────────────────────────────────────────────────────┘

オペレーターサブタスクが失敗すると、Flink は新しい実行試行を作成し、完了した checkpoint から state を復元します。TaskManager が失われると、Flink は置換容量を取得し、そのタスクを再スケジュールします。したがって、永続的な checkpoint storage は一時的なワーカー storage から独立しています。

Flink Client とコマンドラインインターフェイス #

Flink client はデプロイメントの送信および制御側です。Flink 設定を読み込み、アプリケーション JAR をロードし、ジョブを構築または送信し、必要なアーティファクトをアップロードし、クラスターの REST endpoint と通信します。実行中のジョブの一覧表示、停止、キャンセル、state snapshot の取得もできます。detach された送信が成功すると、client はレコード処理経路の一部ではありません。JobManager がジョブを調整し、TaskManager がレコードを処理します。

• session mode では、client がアプリケーションの main メソッドをローカルで呼び出して JobGraph をローカルに構築し、コンパイル済みグラフ表現を既存クラスターへ送信します。
  • main はクラスター内で再度呼び出されることはありませんが、オペレーターロジック（ラムダ）は各 TaskManager で実行されます。
• application mode では、client がユーザーの job jar を HDFS の staging dir に送信し、YARN AM に JobManager の実行を要求します。
  • main は JobManager プロセス内で実行されます。

一般的なコマンド形式は次のとおりです。

./bin/flink <action> [Flink options] <application.jar> [application arguments]

JAR より前のオプションは Flink を設定します。JAR より後の引数はアプリケーションの main メソッドへ渡されます。正確なオプションはインストールされた Flink のバージョンによって異なるため、ローカル client が信頼できる参照先です。

./bin/flink --help
./bin/flink run --help
./bin/flink stop --help
./bin/flink savepoint --help

Job の送信 #

次の例は、既存クラスターへ Java ジョブを送信します。

./bin/flink run \
  --target remote \
  --jobmanager jobmanager.example.com:8081 \
  --detached \
  --parallelism 8 \
  --class com.example.OrderPipeline \
  order-pipeline.jar \
  --input-topic orders \
  --output-topic customer-totals

頻繁に使用される run オプションは次のとおりです。

--detached がない場合、client は接続されたままジョブを待機します。--detached を指定すると、新しい JobID を表示し、送信に成功した後で終了します。実行中のジョブはクラスターが所有するため、detach された client を終了しても session cluster のジョブは終了しません。

-D では任意の適用可能な設定キーを設定できます。たとえば次のようになります。

./bin/flink run \
  -Dpipeline.max-parallelism=128 \
  -Dexecution.checkpointing.interval=30s \
  --detached \
  order-pipeline.jar

既存の session cluster へ送信する場合、変更できるのは execution-level configuration だけです。クラスターのプロセス設定はクラスター起動時に固定されています。

Deployment Target の選択 #

Target は、既存クラスターへ接続するか、application-scope のクラスターを作成するかを client に伝えます。

主要な target の例は次のとおりです。

# Local MiniCluster
./bin/flink run --target local order-pipeline.jar

# Existing standalone cluster
./bin/flink run \
  --target remote \
  --jobmanager localhost:8081 \
  order-pipeline.jar

# Existing YARN session
./bin/flink run \
  --target yarn-session \
  -Dyarn.application.id=application_123_001 \
  order-pipeline.jar

# New YARN application cluster
./bin/flink run \
  --target yarn-application \
  -Dyarn.application.name=order-pipeline \
  order-pipeline.jar

# Existing Kubernetes session
./bin/flink run \
  --target kubernetes-session \
  -Dkubernetes.cluster-id=flink-session \
  order-pipeline.jar

# New Kubernetes application cluster
./bin/flink run \
  --target kubernetes-application \
  -Dkubernetes.cluster-id=order-pipeline \
  -Dkubernetes.container.image.ref=example/order-pipeline:1.0 \
  local:///opt/flink/usrlib/order-pipeline.jar

Kubernetes application mode では、local:/// はコンテナイメージ内にすでに存在する JAR を示します。設定された artifact management support や、サポートされるリモートファイルシステムに保存された JAR も使用できます。YARN と Kubernetes では、これらの CLI 引数に加えて、各プラットフォーム固有の設定と credential が必要です。

Job の調査と制御 #

info はジョブを送信せずにアプリケーションプランを構築します。

./bin/flink info \
  --class com.example.OrderPipeline \
  --parallelism 8 \
  order-pipeline.jar

list を使用して、選択したクラスターが認識しているジョブを調査します。

./bin/flink list
./bin/flink list --running
./bin/flink list --all

# Select a particular remote cluster.
./bin/flink list \
  --target remote \
  --jobmanager jobmanager.example.com:8081

run と list が表示する JobID は、ライフサイクルコマンドでジョブを識別します。Savepoint は明示的に起動できます。

./bin/flink savepoint \
  <JOB_ID> \
  s3://flink-state/savepoints

# Return after Flink accepts the savepoint request.
./bin/flink savepoint \
  <JOB_ID> \
  s3://flink-state/savepoints \
  -detached

execution.checkpointing.savepoint-dir が設定されている場合、target directory は省略できます。Savepoint は Flink を通じて dispose することもでき、その data と metadata の両方をクリーンアップできます。

./bin/flink savepoint \
  --dispose s3://flink-state/savepoints/savepoint-abcd

最近の Flink client では、設定された backend が通常 incremental checkpoint を作成する場合でも、full checkpoint を含めて checkpoint を手動で起動できます。

./bin/flink checkpoint <JOB_ID>
./bin/flink checkpoint <JOB_ID> --full

--fromSavepoint を使用して、savepoint から置換ジョブを復元します。

./bin/flink run \
  --detached \
  --fromSavepoint s3://flink-state/savepoints/savepoint-abcd \
  order-pipeline.jar

デフォルトでは、savepoint 内のすべての state entry が新しいジョブ内のオペレーターに対応している必要があります。--allowNonRestoredState を指定すると、対応しない state を破棄できます。これは、オペレーターを意図的に削除した後に必要になる場合があります。

./bin/flink run \
  --fromSavepoint s3://flink-state/savepoints/savepoint-abcd \
  --allowNonRestoredState \
  order-pipeline.jar

stop は、最終 savepoint を作成してストリーミングジョブを制御された形で終了する場合に使用します。

./bin/flink stop \
  --savepointPath s3://flink-state/savepoints \
  <JOB_ID>

--drain を追加すると、最後の snapshot 前に最終最大 watermark が出力され、event-time timer と window が完了できるようになります。

./bin/flink stop \
  --drain \
  --savepointPath s3://flink-state/savepoints \
  <JOB_ID>

Drain はパイプラインを恒久的に終了するときに適しています。後でジョブを再開する場合は避けてください。snapshot 前にすべての event-time timer を発火させると、復元されたパイプラインのセマンティクスが変わる可能性があります。

最終 savepoint なしでジョブを直ちに終了する必要がある場合は、cancel を使用します。

./bin/flink cancel <JOB_ID>

クラスター選択オプションはライフサイクルコマンドにも適用されます。たとえば、YARN application は -Dyarn.application.id、Kubernetes cluster は -Dkubernetes.cluster-id で選択できます。これは client がローカル設定から意図したクラスターを検出できない場合に必要です。

Flink と YARN の対応 #

YARN デプロイメントでは、コンポーネントの対応は次のとおりです。

起動とスケジューリングの手順は次のとおりです。

1. Flink client がアーティファクトを staging し、YARN application を送信します。
2. YARN の ResourceManager が ApplicationMaster container を起動する NodeManager を選択します。
3. AM が Dispatcher、Flink ResourceManager、REST service、および、ジョブが送信された場合はその JobMaster を起動します。
4. JobMaster が実行グラフに必要な slot を決定します。
5. Flink ResourceManager が、通常の ApplicationMaster protocol を通じて、その要求を YARN container request に変換します。
6. YARN NodeManager が割り当てられた container 内で TaskManager process を起動します。
7. TaskManager が登録し、task slot を提供し、JobMaster からオペレーターサブタスクを受け取ります。

YARN が理解するのは application、queue、node、container、CPU、memory、process health です。Flink operator、Kafka partition、task slot、watermark、checkpoint は理解しません。Flink は、YARN が提供する resource 内でその処理をスケジュールします。

障害処理も同じ境界に従います。

• Task failure は JobMaster と Flink の restart strategy が処理します。
• TaskManager または NodeManager の喪失により、影響を受けた task が replacement capacity 上へ復元されます。
• AM failure は、その control-plane process incarnation を終了させます。YARN は新しい application attempt を開始できますが、完全な job recovery には Flink HA metadata と外部保存された checkpoint も必要です。
• YARN ResourceManager の failover は YARN HA が処理し、その後 Flink AM が通常の YARN protocol を通じて再登録します。

これは YARN 上の Spark に似ていますが、Flink には追加の control-plane layer があります。Spark executor は TaskManager とおおむね対応し、Spark driver は per-job JobMaster に最も近い役割です。Dispatcher と Flink ResourceManager は、それらの JobMaster の周囲で job admission と worker resource を管理します。

Kafka Connector の追加 #

Flink の Kafka connector は core runtime とは別に配布されます。Maven プロジェクトでは、Flink API に加えて connector が必要です。

<properties>
  <flink.version>YOUR_FLINK_VERSION</flink.version>
  <kafka.connector.version>COMPATIBLE_CONNECTOR_VERSION</kafka.connector.version>
</properties>

<dependencies>
  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-streaming-java</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
  </dependency>

  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-clients</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
  </dependency>

  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-kafka</artifactId>
    <version>${kafka.connector.version}</version>
  </dependency>
</dependencies>

Flink Kafka connector documentation に互換性があると記載されたバージョンを使用します。Connector release には、Flink runtime とは独立したバージョンライフサイクルがある場合があります。バージョン文字列が似ているという理由だけで、2 つの artifact に互換性があると想定しないでください。

クラスターへ送信するときは、connector を application とともにパッケージするか、クラスターの依存関係管理規約に従ってインストールします。Core Flink dependency は通常クラスターから提供され、provided として指定されます。Connector のパッケージ方法はクラスターの管理方法に依存します。

Kafka Topic の消費 #

現代的な DataStream connector は KafkaSource を使用します。次の例は orders topic から UTF-8 文字列を読み取ります。

import java.time.Duration;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public final class OrderPipeline {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(30_000, CheckpointingMode.EXACTLY_ONCE);

        KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("kafka:9092")
                .setTopics("orders")
                .setGroupId("order-pipeline")
                .setStartingOffsets(OffsetsInitializer.committedOffsets())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStream<String> orders = env.fromSource(
                source,
                WatermarkStrategy.noWatermarks(),
                "orders-source"
        );

        orders
                .filter(order -> !order.isBlank())
                .name("discard-empty-orders")
                .print();

        env.execute("order-pipeline");
    }
}

重要な source 設定は次のとおりです。

• bootstrap servers: 検出に使用する初期 Kafka broker
• topics or topic pattern: Flink が検出する入力 partition
• group ID: Kafka consumer group の識別子
• starting offsets: 復元された Flink state がないジョブの開始位置
• deserializer: Kafka record を application object に変換する方法

OffsetsInitializer.committedOffsets() には partition の committed offset が必要です。その overload では、commit 済みの位置がない partition に対する reset strategy を指定できます。他の initializer では、最初の offset、最後の offset、timestamp、明示的な offset から開始できます。

.setStartingOffsets(OffsetsInitializer.earliest())

Starting-offset policy は、利用可能な Flink state が復元されない場合にのみ重要です。Checkpoint または savepoint から復旧すると、その Flink snapshot に保存された position が優先されます。

Key、Header、Metadata の Deserialize #

SimpleStringSchema は record value だけを受け取ります。Kafka key、header、timestamp、partition number、offset が必要なアプリケーションでは、代わりに KafkaRecordDeserializationSchema<T> を実装します。

KafkaSource<OrderEvent> source = KafkaSource.<OrderEvent>builder()
        .setBootstrapServers("kafka:9092")
        .setTopics("orders")
        .setGroupId("order-pipeline")
        .setStartingOffsets(OffsetsInitializer.earliest())
        .setDeserializer(new OrderEventDeserializationSchema())
        .build();

Deserialize は決定的に保ちます。無効なレコードには、ジョブを失敗させる、または元の topic、partition、offset、error とともに dead-letter stream へ送るなど、明示的な policy を適用します。形式不正の business event を黙って破棄すると、復旧と監査が困難になります。

新しい Partition の検出 #

ジョブの実行中に Kafka topic が partition を追加することがあります。Connector は定期的にそれらを検出できます。

.setProperty("partition.discovery.interval.ms", "60000")

新しい partition には、定義された starting-offset policy が必要です。Topic の partition 数を増やすと producer の key-to-partition assignment も変わり、Flink 外部の ordering assumption に影響する可能性があります。

Event Time でイベントを処理する #

各 Kafka value が次の record に解析されるとします。

public record Purchase(
        String customerId,
        long amountInCents,
        long occurredAt
) {
}

Flink は customer ごとの 1 分間の合計を計算できます。

import java.time.Duration;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

WatermarkStrategy<Purchase> watermarks =
        WatermarkStrategy
            .<Purchase>forBoundedOutOfOrderness(Duration.ofSeconds(10))
            .withTimestampAssigner(
                (purchase, previousTimestamp) -> purchase.occurredAt()
            )
            .withIdleness(Duration.ofMinutes(1));

DataStream<Purchase> purchases = rawOrders
        .map(OrderPipeline::parsePurchase)
        .assignTimestampsAndWatermarks(watermarks);

DataStream<Purchase> minuteTotals = purchases
        .keyBy(Purchase::customerId)
        .window(TumblingEventTimeWindows.of(Time.minutes(1)))
        .reduce((left, right) -> new Purchase(
                left.customerId(),
                left.amountInCents() + right.amountInCents(),
                Math.max(left.occurredAt(), right.occurredAt())
        ));

withIdleness は、アイドル状態の Kafka partition が結合された watermark を無期限に遅らせるのを防ぎます。期間は任意に決めるのではなく、観測されたトラフィックパターンから選択します。短すぎる値は、一時的に静かな partition を早期に完了したように扱う可能性があります。

Kafka は 1 つの partition 内の順序だけを保持し、topic 全体の順序は保持しません。Flink の event-time model は partition 間のインターリーブを処理しますが、許容する lateness の量と、非常に遅いイベントへの対応をアプリケーションが定義する必要があります。

結果を Kafka に公開する #

KafkaSink は DataStream を Kafka に書き込みます。

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;

KafkaSink<String> sink = KafkaSink.<String>builder()
        .setBootstrapServers("kafka:9092")
        .setRecordSerializer(
            KafkaRecordSerializationSchema.builder()
                .setTopic("order-totals")
                .setValueSerializationSchema(new SimpleStringSchema())
                .build()
        )
        .setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE)
        .build();

results.sinkTo(sink).name("order-totals-sink");

出力 topic、key、header、timestamp が record によって変わる場合は、KafkaRecordSerializationSchema<T> を直接使用します。

Sink は 3 つの配送保証をサポートします。

• NONE: checkpointing の一部として Kafka acknowledgment を待たない
• AT_LEAST_ONCE: acknowledgment 済みレコードが失われないことを保証しますが、復旧時に重複が発生する可能性がある
• EXACTLY_ONCE: Flink checkpoint と調整された Kafka transaction を使用

Exactly-once output には checkpointing が必要です。

KafkaSink<String> sink = KafkaSink.<String>builder()
        .setBootstrapServers("kafka:9092")
        .setRecordSerializer(
            KafkaRecordSerializationSchema.builder()
                .setTopic("order-totals")
                .setValueSerializationSchema(new SimpleStringSchema())
                .build()
        )
        .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
        .setTransactionalIdPrefix("order-pipeline-")
        .build();

同時に実行される各ジョブには、一意の transactional ID prefix が必要です。Prefix を再利用すると、ある producer が別の producer を fence する可能性があります。未 commit の transactional output を参照してはいけない Kafka consumer では、isolation.level=read_committed を使用します。

Exactly-once は sink option だけではなく、end-to-end property です。次の要素が必要です。

• replay 可能な入力
• 決定的、または正しく checkpoint される処理
• 成功した Flink checkpoint
• transactional または idempotent な destination
• committed result を適切に参照する consumer

map または process function 内の外部 side effect、たとえば HTTP request の送信は、Flink が checkpoint を復元したときに自動的にロールバックされません。Idempotency key、transactional integration、または Kafka に intent を公開して別システムが実行する設計を使用します。

Offset と Checkpoint の連携 #

Kafka consumer group は通常、committed offset によって進捗を追跡します。Flink も checkpoint に source position を保存します。これらの仕組みは関連していますが、役割は異なります。

1. Source が割り当てられた Kafka partition からレコードを読み取ります。
2. Flink checkpoint barrier がジョブを通過します。
3. オペレーターが source position と整合する形で state を snapshot します。
4. 必要なすべての snapshot と sink action が成功すると checkpoint が完了します。
5. 障害時に Flink がその完了済み checkpoint から operator state と Kafka position を復元します。

Kafka committed offset は可視性や initial start policy には役立ちますが、Flink checkpoint state の代替ではありません。Flink state を削除して consumer-group offset から再起動すると、source progress と operator state の正確な関係が失われる可能性があります。

Checkpoint storage は永続的で、クラスターからアクセスできる必要があります。一時ワーカー上のローカルファイルシステムパスは本番には不十分です。Recovery と deployment strategy に従って checkpoint を保持するか、savepoint を取得します。

Flink SQL で Kafka を使用する #

Flink の Table API と SQL connector は、Kafka topic を dynamic table として公開します。次のテーブルは JSON order event を読み取ります。

CREATE TABLE orders (
  order_id STRING,
  customer_id STRING,
  amount DECIMAL(12, 2),
  occurred_at TIMESTAMP(3),
  WATERMARK FOR occurred_at AS occurred_at - INTERVAL '10' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'kafka:9092',
  'properties.group.id' = 'order-sql',
  'scan.startup.mode' = 'group-offsets',
  'format' = 'json',
  'json.fail-on-missing-field' = 'false',
  'json.ignore-parse-errors' = 'false'
);

2 つ目の Kafka table は aggregate を受け取れます。

CREATE TABLE customer_totals (
  customer_id STRING,
  window_start TIMESTAMP(3),
  window_end TIMESTAMP(3),
  total_amount DECIMAL(18, 2),
  PRIMARY KEY (customer_id, window_start) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'customer-totals',
  'properties.bootstrap.servers' = 'kafka:9092',
  'key.format' = 'json',
  'value.format' = 'json'
);

Query はイベントを event-time window にグループ化します。

INSERT INTO customer_totals
SELECT
  customer_id,
  window_start,
  window_end,
  SUM(amount)
FROM TABLE(
  TUMBLE(TABLE orders, DESCRIPTOR(occurred_at), INTERVAL '1' MINUTE)
)
GROUP BY customer_id, window_start, window_end;

Append-only record には通常の kafka connector を使用します。Primary key で識別される更新または削除を table が生成する場合は upsert-kafka を使用します。通常の Kafka topic が自動的に upsert table になるわけではありません。Key format、value format、changelog semantics、下流 consumer が一致する必要があります。

SQL connector artifact も Flink cluster から利用可能でなければなりません。正確なインストール方法は、SQL client、application JAR、managed Flink service のいずれを使用するかによって異なります。

Partitioning と Parallelism #

Kafka partition は、1 つの topic を同時に読み取れる source subtask 数の最大値を決定します。12 partition と source parallelism 20 の場合、少なくとも 8 つの source subtask には読み取る partition がありません。Source parallelism が 4 の場合、各 subtask は複数の partition を読み取ります。

TaskManager 数は Kafka partition 数とは独立しています。関連する制限は worker-process 数ではなく、Kafka source parallelism です。たとえば、以下はいずれも 12-partition topic に対して 12 source slot を提供できます。

3 TaskManagers x 4 slots
6 TaskManagers x 2 slots
12 TaskManagers x 1 slot

Parsing、非同期 API 呼び出し、keyed processing、sink に、Kafka の読み取りより大きな capacity が必要な場合、ジョブには source partition より多くの TaskManager または下流 parallelism が存在することもあります。Source parallelism を検出された partition 数以上に増やすと、一部の source subtask は idle になりますが、ジョブの他の部分に対応する上限を課すわけではありません。

すべての operator が同じ parallelism である必要はありません。keyBy の後、Flink は論理 key に従ってレコードを再パーティション化し、下流 operator を異なる parallelism で実行できます。

Partition 数と parallelism は次の要素を使って選択します。

• peak input および output throughput
• record size と serialization cost
• Kafka key と Flink key の skew
• key あたりの state size
• checkpoint duration
• broker と network capacity
• 目的とする rescaling range

Partition を増やすと潜在的な並行性は増えますが、file、connection、metadata、checkpoint work も増えます。Hot key は解決しません。1 つの Flink key のすべてのレコードは、アプリケーションが aggregation strategy を変更しない限り、1 つの keyed subtask に送られます。

Schema Management #

JSON は例に便利ですが、それだけでは強力な schema compatibility を提供しません。本番パイプラインでは、schema registry とともに Avro、Protobuf、JSON Schema を使用することが一般的です。

Schema evolution を計画します。

• 安定した field name と意味を割り当てる
• 必須 field と optional field を区別する
• field の追加時には互換性のある default を使用する
• 新しい意味のために既存 field を再利用しない
• consumer が必要とする可能性がある場合は unknown field を保持する
• 古い record を新しい application version に対してテストする

Kafka retention により、schema compatibility は特に重要です。復元または再処理するジョブが数か月前に生成されたイベントを読み取る可能性があります。

Deserialize failure には、source record を診断するのに十分な context を含めます。組織が dead-letter topic を許可している場合は、元の topic、partition、offset、key、schema identifier、安全な error description を含めます。機密 payload を制限のない log にコピーしないでください。

Security Configuration #

Connector は TLS と SASL の Kafka client property を受け付けます。デプロイメントでは次のような property を指定できます。

.setProperty("security.protocol", "SASL_SSL")
.setProperty("sasl.mechanism", "SCRAM-SHA-512")
.setProperty(
    "sasl.jaas.config",
    System.getenv("KAFKA_SASL_JAAS_CONFIG")
)

Password や JAAS credential を source code にハードコードしないでください。Platform の secret mechanism から注入し、job configuration と log へのアクセスを制限します。

Kafka ACL には最小権限を適用します。Source job には通常、topic description と read access、および consumer group の権限が必要です。Sink には write access が必要であり、exactly-once delivery では transactional producer に必要な権限も必要です。

暗号化された broker connection を使用し、broker certificate を検証し、近道として hostname verification を無効化しないでください。

Pipeline の運用 #

正常な Kafka-Flink pipeline では、両システムを可視化する必要があります。

少なくとも次を監視します。

• input と output の record および byte
• partition ごとの Kafka consumer lag
• Flink subtask の busy、idle、backpressured time
• checkpoint duration、size、failure、alignment time
• restart count と failure cause
• watermark progression と idle partition
• late または invalid record count
• state size と増加量
• Kafka produce latency と error
• exactly-once sink の transaction timeout

Consumer lag だけでは不十分です。Lag が低くても、ジョブが不正な結果を生成していたり、checkpoint に繰り返し失敗していたり、state を無制限に蓄積していたりする可能性があります。逆に、意図的な bounded replay は lag が高くても正常に動作していることがあります。

Stateful production job には安定した operator UID を設定します。

DataStream<OrderTotal> totals = purchases
        .keyBy(Purchase::customerId)
        .process(new CustomerTotalFunction())
        .uid("customer-total")
        .name("customer-total");

Flink は operator identity を使用して、upgrade 中に saved state を対応付けます。明示的な UID によって、無害な code-graph change が state restoration を妨げるリスクを減らせます。UID を不用意に変更または再利用しないでください。

本番前に recovery をテストします。有用なテストでは、record が流れている間に task を停止し、復元を待ち、最終 output について duplicate、loss、ordering assumption を検証します。Savepoint upgrade、rescaling、Kafka broker interruption、malformed record、slow destination もテストします。

よくある落とし穴 #

Kafka の順序をグローバルだと仮定する #

Kafka が順序を保証するのは partition 内だけです。ある entity のすべての event を順序どおりに維持する必要がある場合、producer は安定した key を使用して、それらの event が同じ partition に届くようにします。後段の Flink keyBy で event を再グループ化することはできますが、存在しなかったグローバル順序を復元することはできません。

Processing Guarantee を混同する #

EXACTLY_ONCE checkpoint mode によって、任意の外部 call が exactly-once になるわけではありません。障害と replay の動作を通じて、すべての side effect を追跡します。

間違った Offset から開始する #

earliest、latest、committed offset、timestamp、restored state は異なる意味を持ちます。インシデントが発生する前に、restart と full-replay の手順を明示します。

Watermark Idleness を無視する #

静かな Kafka partition 1 つによって、下流の event-time watermark が停止する可能性があります。Partition が正当にレコードを受信しないことがある場合は idleness を設定します。

Unbounded な Key または State Model を使用する #

Request ID のような key は、数百万の小さな state entry を作成する可能性があります。State には cleanup mechanism が必要です。Window、timer、state time-to-live、または正確性への影響を理解した別の business rule を使用します。

Dead-Letter Topic を解決策として扱う #

Dead-letter topic は問題のある record を保持しますが、問題を解決するものではありません。所有者、alerting、retention、replay、privacy control を定義します。

実践的な設計チェックリスト #

Kafka に接続された Flink job をデプロイする前に、次の質問へ回答します。

1. 必要な ordering を維持する Kafka key は何か。
2. Flink state が存在しない場合、新しい deployment はどこから開始するか。
3. 計算を駆動する event timestamp は何か。
4. Event はどの程度遅れて到着する可能性があり、その上限を超えた場合に何が起こるか。
5. 各 key が保持する state は何で、どのように cleanup するか。
6. 各 output に必要な delivery guarantee は何か。
7. すべての外部 side effect は replay-safe か。
8. Checkpoint はどこに保存し、どの期間保持するか。
9. Job をどのように upgrade、rollback、rescale するか。
10. Schema をどのように進化させ、malformed record をどのように処理するか。
11. Lag、backpressure、checkpoint failure を示す metric と alert は何か。
12. 現実的な traffic で failure と recovery をテストしたか。

Kafka は Flink に永続的で replay 可能な event log を提供します。Flink は distributed state、event-time processing、recovery、豊富な transformation を追加します。両者を組み合わせることで、streaming application の強固な基盤を形成できます。ただし、正確性は offset、checkpoint、transaction、schema、partitioning、および pipeline 外部のすべての side effect に関する意図的な選択に依存します。