メインコンテンツまでスキップ
に最終更新

Flink Doris Connector

Flink Doris Connector を使用すると、Flink を介して Doris クラスタのデータを読み書きできます。また、Flink CDC と連携し、MySQL などの上流データベース全体を簡単に同期できます。

機能​

機能説明主な用途
Doris からのデータの読み取りFE を経由せず、複数の BE から並列に読み取ります。Thrift および Arrow Flight SQL プロトコルに対応していますDoris テーブルを Flink ジョブの入力として使用する、データを一括エクスポートする
Doris の変更ログを使った増分読み取りテーブルの行単位の変更（挿入、更新、削除）を継続的に取得します。スナップショットからの開始も可能です。Connector 26.4.0+ と Doris 5.0.0+ が必要ですDoris テーブルの変更を下流のシステムへリアルタイムで同期する
Doris へのデータの書き込みFlink でデータをバッチにまとめ、Stream Load を介して書き込みます。ストリーミング書き込み（Checkpoint ベース、Exactly-Once）、バッチ書き込み、S3 TVF による書き込みに対応していますリアルタイムのデータ取り込み、CDC による変更の取り込み、列の部分更新
Lookup Join上流のレコードをバッチにまとめ、Doris のディメンションテーブルに非同期でクエリを実行しますストリームにディメンションデータを付加する
データベース全体の同期Flink CDC と連携し、MySQL、Oracle、PostgreSQL、SQLServer、DB2、MongoDB のデータベース全体を単一のコマンドで同期します。テーブルの作成と DDL の同期も自動で行います業務データベースを Doris にリアルタイムで同期する

すべての機能を FlinkSQL から利用できます。読み取りと書き込みは DataStream API からも利用できます。接続設定については接続オプションと TLS、Doris と Flink のデータ型の対応についてはデータ型のマッピングを参照してください。

仕組み​

FlinkConnectorPrinciples-JDBC-Doris

  • 読み取り: Connector は FE からクエリプランの Tablet 情報を取得し、BE から直接、並列にデータを読み取ります。データは FE を経由しないため、JDBC を介して単一の FE ノードで読み書きを直列に処理する際のボトルネックを回避できます。詳しくは読み取りの仕組みを参照してください。
  • 書き込み: Flink のメモリ上でデータをバッチにまとめ、Stream Load を介して Doris に一括データロードします。デフォルトでは、書き込みは Flink の Checkpoint に従い、二相コミットによって Exactly-Once を保証します。Checkpoint に依存しないバッチ書き込みモードも利用できます。詳しくは書き込みモードを参照してください。
  • Lookup Join: レコードごとにポイントクエリを実行する代わりに、上流のレコードをバッチにまとめ、Doris のディメンションテーブルに非同期でクエリを実行します。
  • データベース全体の同期: Flink CDC で上流データベースの変更を読み取り、Doris Sink を介して書き込みます。Doris Sink は対象テーブルの作成と Schema Changes の同期を自動で行います。

バージョンについて​

Connector のバージョンFlink のバージョンDoris のバージョンJava のバージョンScala のバージョン
1.0.31.11,1.12,1.13,1.140.15+82.11,2.12
1.1.11.141.0+82.11,2.12
1.2.11.151.0+8-
1.3.01.161.0+8-
1.4.01.15 - 1.171.0+8-
1.5.21.15 - 1.181.0+8-
1.6.11.15 - 1.191.0+8-
24.0.11.15 - 1.201.0+8-
24.1.01.15 - 1.201.0+8-
25.0.01.15 - 1.201.0+8-
25.1.01.15 - 1.201.0+8-
26.0.01.15 - 1.20,2.0 - 2.21.0+8(1.x),17(2.x)-
26.1.11.15 - 1.20,2.0 - 2.21.0+8(1.x),17(2.x)-
26.2.01.15 - 1.20,2.0 - 2.21.0+8(1.x),17(2.x)-
26.3.01.15 - 1.20,2.0 - 2.31.0+8(1.x),17(2.x)-

インストール​

Jar パッケージと Maven 依存関係のどちらの方法も利用できます。

Jar パッケージを使用する方法​

Doris のダウンロードページから対応するバージョンの Flink Doris Connector の Jar パッケージをダウンロードし、Flink の classpath にコピーします。

  • Standalone モード: Jar ファイルを lib/ ディレクトリに配置します。
  • Yarn クラスタモード: Jar ファイルを事前デプロイ用パッケージに含めます。

Maven 依存関係を使用する方法​

プロジェクトの pom.xml に次の依存関係を追加します。

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

例:

<dependency>
    <groupId>org.apache.doris</groupId>
    <artifactId>flink-doris-connector-1.16</artifactId>
    <version>26.3.0</version>
</dependency>

次のステップ​

  • クイックスタートに従い、FlinkSQL で Doris テーブルを読み書きします。
  • エラーが発生した場合は、まず FAQ を確認してください。
Report issue