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

Flink

Flink Doris Connectorを使用すると、Flinkが生成したデータ（FlinkがKafkaやMySQLから読み取ったデータなど）をリアルタイムでDorisに取り込めます。リアルタイムのデータ取り込みやストリーミングETLに適しています。

適用シナリオ​

シナリオ説明
リアルタイムのデータ取り込みKafkaやPulsarなどのメッセージキューからDorisにデータをリアルタイムで書き込む
データベースの同期Flink CDCを使用して、MySQLやOracleなどのデータベースからDorisにデータを同期する
ストリーミングETLFlinkでリアルタイム処理を行い、結果をDorisに書き込む

制限事項​

  • デプロイ済みのFlinkクラスターが必要です。
  • 対応するバージョンのFlink Doris ConnectorをFlinkにデプロイする必要があります。

手順​

Flinkを使用したデータ取り込みの詳細な手順については、Flink-Doris-Connectorを参照してください。以下の最小構成の例では、Flinkによるデータ取り込みの手順を示します。

全体の手順は次のとおりです。

  1. Dorisにターゲットテーブルを作成する
  2. FlinkでFlinkSQLを使用してデータを書き込む
  3. Dorisでデータが正常に取り込まれたことを確認する

ステップ1：Dorisにテーブルを作成する​

Flinkからのデータを受け取るターゲットテーブルstudentsをDorisに作成します。

CREATE TABLE `students` (
    `id` INT NULL,
    `name` VARCHAR(256) NULL,
    `age` INT NULL
) ENGINE=OLAP
UNIQUE KEY(`id`)
COMMENT 'OLAP'
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
    "replication_allocation" = "tag.location.default: 1"
);

bin/sql-client.shを実行してFlinkSQLコンソールを開き、次のステートメントを実行してシンクテーブルを作成し、データを書き込みます。

CREATE TABLE student_sink (
    id INT,
    name STRING,
    age INT
    )
    WITH (
      'connector' = 'doris',
      'fenodes' = '10.16.10.6:28737',
      'table.identifier' = 'test.students',
      'username' = 'root',
      'password' = '',
      'sink.label-prefix' = 'doris_label'
);

INSERT INTO student_sink values(1,'zhangsan',123)

主なパラメーターは次のとおりです。

パラメーター説明
connectordorisに固定します。Flink Doris Connectorを使用することを示します
fenodesDoris FEのHTTPアドレス。形式はhost:http_portです
table.identifierターゲットテーブルの識別子。形式はdatabase.tableです
usernameDorisのログインユーザー名
passwordDorisのログインパスワード
sink.label-prefixStream Loadの取り込みタスクに使用するLabelのプレフィックス。グローバルに一意である必要があります

ステップ3：取り込んだデータを確認する​

Dorisでターゲットテーブルをクエリし、データが正常に取り込まれたことを確認します。

select * from test.students;
+------+----------+------+
| id   | name     | age  |
+------+----------+------+
|  1   | zhangsan |  123 |
+------+----------+------+

よくある質問​

Q1：Flink Doris Connectorはどのデータソースに対応していますか？

理論上、Flinkが対応するすべてのデータソース（Kafka、MySQL CDC、ファイルシステム、メッセージキューなど）を上流のソースとして使用できます。Flinkでデータを処理してから、Dorisに書き込めます。

Q2：sink.label-prefixを一意にする必要があるのはなぜですか？

Flink Doris Connectorは、DorisのStream Loadを使用してデータを取り込みます。Exactly-Onceセマンティクスを保証するには、各トランザクションに一意のLabelが必要です。Labelが重複すると、データ取り込み時に競合が発生します。

Q3：fenodesにはどのポートを指定しますか？

fenodesには、Doris FEのHTTPポート（デフォルトは8030）を指定します。MySQLプロトコルのポート（デフォルトは9030）ではありません。

Q4：MySQLやOracleなどのデータベースからDorisにデータを同期するにはどうすればよいですか？

Flink CDCと組み合わせて使用できます。詳細については、Flink-Doris-Connectorのドキュメントを参照してください。

Report issue