Realtime Compute for Apache Flink:SelectDB

全般

ソーステーブル

結果テーブル

書き込みモードは、配信保証とフラッシュ動作に影響します。一貫性の要件に基づいて選択してください。

**sink.properties.* の例：**

CSV フォーマット：

'sink.properties.column_separator' = ','
-- 値にカンマが含まれる可能性がある場合は、印刷不可能な区切り文字を使用します：
-- 'sink.properties.column_separator' = '\x01'

JSON フォーマット：

'sink.properties.format'            = 'json',
'sink.properties.read_json_by_line' = 'true'
-- または： 'sink.properties.strip_outer_array' = 'true'

ディメンションテーブル

単一テーブルの同期

単一の MySQL テーブルを SelectDB に同期します。宛先テーブルは存在しない場合、自動的に作成されます。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  # (任意) 最初に既存の完全データを同期し、次に増分データを同期します。
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

pipeline:
  name: MySQL to SelectDB Pipeline

データベース全体の同期

MySQL データベース内のすべてのテーブルを SelectDB に同期します。ダウンストリームのデータベース名とテーブル名は、アップストリームのものと同じです。宛先テーブルは存在しない場合、自動的に作成されます。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  # test_db データベース内のすべてのテーブルに一致します。
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial
  # (任意) ジョブを再起動せずに、増分フェーズ中に作成されたテーブルを同期します。
  scan.binlog.newly-added-table.enabled: true

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}
  # (任意) バッチ書き込みのフラッシュ間隔。小規模なワークロードでは、データがバッファーに長時間保持されないように、この値を小さくします。
  sink.buffer-flush.interval: 10s

pipeline:
  name: MySQL to SelectDB Pipeline

データベース全体の同期からテーブルを除外

一時テーブルやログテーブルなど、同期したくないテーブルをスキップします。ルートモジュールがない場合、データはソーステーブルと同じ名前の宛先テーブルに書き込まれます。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  tables: test_db.\.*
  # この正規表現に一致するテーブルは同期されません。複数の正規表現はカンマ (,) で区切ります。
  tables.exclude: test_db.tmp_\.*, test_db.log_\.*
  server-id: 5401-5499
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

pipeline:
  name: MySQL to SelectDB Pipeline

指定されたデータベースとテーブルへの同期

ダウンストリームのデータベース名またはテーブル名がアップストリームのものと異なる場合は、ルートモジュールを使用して宛先の名前を変更します。次の例では、test_db のテーブルを ods_db データベースに同期し、各テーブル名に ods_ プレフィックスを追加します。たとえば、test_db.orders は ods_db.ods_orders に同期されます。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  tables: test_db.\.*
  server-id: 5401-5499
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

route:
  # <> は、一致したソーステーブル名に置き換えられるプレースホルダーです。
  - source-table: test_db.\.*
    sink-table: ods_db.ods_<>
    replace-symbol: <>

pipeline:
  name: MySQL to SelectDB Pipeline

シャードテーブルのマージ

同じスキーマを共有する複数のシャードテーブルを単一の SelectDB テーブルにマージします。次の例では、変換モジュールを使用してソースデータベースとテーブルを識別する列を追加し、この列とビジネスプライマリキーを複合プライマリキーとして使用して、マージされたデータの出所を追跡できるようにします。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  # order_db_1.orders_1 や order_db_2.orders_2 など、すべてのシャードに一致します。
  tables: order_db_\d+.orders_\d+
  server-id: 5401-5499
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}
  # ダウンストリームのプライマリキーがアップストリームのものと異なる場合は、このパラメーターを false に設定します。そうしないと、古いプライマリキーに対応する行が削除されません。
  sink.ignore.update-before: false

transform:
  - source-table: order_db_\d+.orders_\d+
    # ソースデータベースとテーブルを識別する src_table 列を追加します。
    projection: "*, __schema_name__ || '.' || __table_name__ AS src_table"
    # 複合プライマリキーは、宛先テーブルのキー列と同じでなければなりません。
    primary-keys: order_id, src_table
    description: ソース識別子を追加し、複合プライマリキーを設定します

route:
  # すべてのシャードを単一のテーブルにマージします。
  - source-table: order_db_\d+.orders_\d+
    sink-table: dw_db.merged_orders

pipeline:
  name: MySQL sharding to SelectDB Pipeline

データのフィルタリングと列のプルーニング

変換モジュールを使用して、データをフィルタリングし、列をプルーニングします。フィルター条件はソーステーブルのフィールドで評価され、レコードを同期するかどうかを決定し、プロジェクションはダウンストリームテーブルに書き込む列を決定します。

source:
  type: mysql
  name: MySQL Source
  hostname: <yourHostname>
  port: 3306
  username: ${secret_values.mysql_username}
  password: ${secret_values.mysql_password}
  tables: test_db.test_source_table
  server-id: 5401-5499
  scan.startup.mode: initial

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

transform:
  - source-table: test_db.test_source_table
    # order_status が PAID のレコードのみを同期します。
    filter: "order_status = 'PAID'"
    # 以下の4つの列のみを書き込みます。
    projection: order_id, customer_id, total_amount, created_at
    primary-keys: order_id
    description: 注文ステータスでフィルタリングし、列をプルーニングします

pipeline:
  name: MySQL to SelectDB Pipeline

PostgreSQL データベース全体の同期

PostgreSQL スキーマ内のすべてのテーブルを SelectDB に同期します。開始する前に、PostgreSQL インスタンスで wal_level を logical に設定し、max_replication_slots と max_wal_senders に十分な余裕があることを確認してください。詳細については、「PostgreSQL データベースの設定」をご参照ください。

source:
  type: postgres
  name: PostgreSQL Source
  hostname: <yourHostname>
  port: 5432
  username: ${secret_values.pg_username}
  password: ${secret_values.pg_password}
  # PostgreSQL のテーブル名は database.schema.table フォーマットを使用します。
  tables: test_db.public.\.*
  slot.name: <yourSlotName>
  decoding.plugin.name: pgoutput
  scan.startup.mode: initial
  # (任意) ジョブ開始時にパブリケーションを作成する代わりに、既存のパブリケーションを使用します。
  debezium.publication.autocreate.mode: disabled
  debezium.publication.name: <yourPublicationName>

sink:
  type: doris
  name: SelectDB Sink
  fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
  jdbc-url: jdbc:mysql://selectdb-cn-****.selectdbfe.rds.aliyuncs.com:9030
  username: ${secret_values.selectdb_username}
  password: ${secret_values.selectdb_password}

pipeline:
  name: PostgreSQL to SelectDB Pipeline