Realtime Compute for Apache Flink：SelectDB

使用方法

1. 單擊JAR包擷取SelectDB自訂連接器（需要為1.15~1.17）。

2. 在Realtime Compute開發控制台上，上傳SelectDB自訂連接器，詳情請參見管理自訂連接器。

3. 在SQL作業中使用SelectDB自訂連接器，connector固定值為doris。

文法結構

CREATE TABLE selectdb_source (
  order_id      BIGINT,
  user_id       BIGINT,
  total_amount  DECIMAL(10, 2),
  order_status  TINYINT,
  create_time   TIMESTAMP(3),
  product_name  STRING COMMENT
) WITH (
  'connector' = 'doris',
  'fenodes' = 'selectdb-cn-*******.selectdbfe.rds.aliyuncs.com:8080',
  'table.identifier' = 'shop_db.orders',
  'username' = 'admin',
  'password' = '****'
);

WITH參數

• 通用

  參數	說明	資料類型	是否必填	預設值	備忘
  connector	表類型。	String	是	無	固定值為doris。
  fenodes	ApsaraDB for SelectDB執行個體的訪問和HTTP協議址地連接埠。	String	是	無	可以從ApsaraDB for SelectDB控制台的執行個體詳情 > 網路資訊中擷取VPC地址（或公網地址）和HTTP協議連接埠。 樣本：selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080。
  jdbc-url	jdbc 串連資訊，	String	否	無	可以從ApsaraDB for SelectDB控制台的執行個體詳情 > 網路資訊中擷取VPC地址（或公網地址）和MySQL協議連接埠。 樣本：jdbc:mysql://selectdb-cn-***.selectdbfe.rds.aliyuncs.com:9030。
  table.identifier	資料庫表名。	String	是	無	樣本：db.tbl。
  username	使用者名稱	String	是	無	如果遺忘密碼，可以從ApsaraDB for SelectDB控制台的執行個體詳情右上方進行重設。
  password	密碼	String	是	無
  doris.request.retries	發送請求的重試次數。	Integer	否	3	無。
  doris.request.connect.timeout	發送請求的連線逾時時間。	Duration	否	30s	無。
  doris.request.read.timeout	發送請求的讀取逾時時間。	Duration	否	30s	無。

• 源表專屬

  參數	說明	資料類型	是否必填	預設值	備忘
  doris.request.query.timeout	查詢逾時時間，預設值為 6 小時	Duration	否	21600s	固定值為doris。
  doris.request.tablet.size	一個 Partition 對應的 Tablet 個數。	Integer	否	1	此數值設定越小，則會產生越多的 Partition。從而提升 Flink 側的並行度，但同時會對資料庫造成更大的壓力。
  doris.batch.size	一次從 BE 讀取資料的最大行數。	Integer	否	4064	增大此數值可減少 Flink 與資料庫之間建立串連的次數。從而減輕網路延遲所帶來的額外時間開銷。
  doris.exec.mem.limit	單個查詢的記憶體限制。	Integer	否	8192mb	預設為 8GB，單位為位元組。
  source.use-flight-sql	是否使用 Arrow Flight SQL 讀取。	Boolean	否	false	無需配置。請直接從ApsaraDB for SelectDB控制台的執行個體詳情 > 網路資訊中單擊開通叢集直連。
  source.flight-sql-port	使用 Arrow Flight SQL 讀取時，FE 的 arrow_flight_sql_port。	Integer	否	-	無。

• 結果表專屬

  參數	說明	資料類型	是否必填	預設值	備忘
  sink.label-prefix	Stream Load 匯入使用的 label 首碼。	String	否	--	多作業情境下要求全域唯一，用來保證 Flink 的 EOS 語義。相同的Label只能匯入一次，確保不重複寫入。
  sink.properties.*	Stream Load 的匯入參數。	String	否	--	CSV 格式配置 'sink.properties.column_separator' = ',', -- 使用逗號分隔 -- 如果資料中可能包含逗號，建議使用不可見字元，如： -- 'sink.properties.column_separator' = '\x01' JSON 格式配置 'sink.properties.format' = 'json', 'sink.properties.read_json_by_line' = 'true' -- 或使用 strip_outer_array
  sink.enable-delete	是否啟用刪除。此選項需要 Doris 表開啟大量刪除功能。	Boolean	否	true	只支援 Unique 模型。
  sink.enable-2pc	是否開啟兩階段交易認可 (2PC)。	Boolean	否	true	保證 Exactly-Once 語義。更多兩階段交易認可請參考明確交易操作。
  sink.buffer-size	寫資料緩衝 buffer 大小。	Integer	否	1MB	單位位元組。不建議修改，預設配置即可。
  sink.buffer-count	寫資料緩衝 buffer 個數。	Integer	否	3	不建議修改，預設配置即可
  sink.max-retries	Commit 失敗後的最大重試次數。	Integer	否	3	無。
  sink.enable.batch-mode	是否使用攢批模式寫入。	Boolean	否	false	開啟後寫入時機不依賴 Checkpoint，通過sink.buffer-flush.max-rows/sink.buffer-flush.max-bytes/sink.buffer-flush.interval參數來控制寫入時機。 同時開啟後將不保證 Exactly-once 語義，但可藉助 Uniq 模型做到等冪。
  sink.flush.queue-size	攢批模式下，緩衝的隊列大小。	Integer	否	2	無。
  sink.buffer-flush.max-rows	攢批模式下，單個批次最多寫入的資料行數。	Integer	否	500000	無。
  sink.buffer-flush.max-bytes	攢批模式下，單個批次最多寫入的位元組數。	Integer	否	100MB	單位位元組。
  sink.buffer-flush.interval	攢批模式下，非同步重新整理緩衝的間隔。	String	否	10s	單位毫秒。
  sink.ignore.update-before	是否忽略 update-before 事件。	Boolean	否	true	無。

• 維表專屬

  參數	說明	資料類型	是否必填	預設值	備忘
  lookup.cache.max-rows	lookup 緩衝的最大行數。	Integer	否	-1	-1預設為不開啟緩衝。
  lookup.cache.ttl	lookup 緩衝的最大時間。	String	否	10s	單位毫秒。
  lookup.max-retries	lookup 查詢失敗後的重試次數	Integer	否	1	無。
  lookup.jdbc.async	是否開啟非同步 lookup。	Boolean	否	false	無。
  lookup.jdbc.read.batch.size	非同步 lookup 下，每次查詢的最大批次大小。	Integer	否	128	無。
  lookup.jdbc.read.batch.queue-size	非同步 lookup 時，中間緩衝隊列的大小。	Integer	否	256	無。
  lookup.jdbc.read.thread-size	每個 task 中 lookup 的 jdbc 線程數。	Integer	否	3	無。

使用樣本

CREATE TEMPORARY TABLE selectdb_source (
  order_id      BIGINT,
  user_id       BIGINT,
  total_amount  DECIMAL(10, 2),
  order_status  TINYINT,
  create_time   TIMESTAMP(3),
  product_name  STRING COMMENT
) WITH (
  'connector' = 'doris',
  'fenodes' = 'selectdb-cn-*******.selectdbfe.rds.aliyuncs.com:8080',
  'table.identifier' = 'shop_db.orders',
  'username' = 'admin',
  'password' = '****'
);

CREATE TEMPORARY TABLE selectdb_source (
  order_id      BIGINT,
  user_id       BIGINT,
  total_amount  DECIMAL(10, 2),
  order_status  TINYINT,
  create_time   TIMESTAMP(3),
  product_name  STRING COMMENT
) WITH (
  'connector' = 'doris',
  'fenodes' = 'selectdb-cn-*******.selectdbfe.rds.aliyuncs.com:8080',
  'table.identifier' = 'shop_db.orders',
  'username' = 'admin',
  'password' = '****',
--  'sink.label-prefix' = 'flink_orders' --相同的Label只能匯入一次，確保不重複寫入。
);

CREATE TEMPORARY TABLE fact_table (
  `id` BIGINT,
  `name` STRING,
  `city` STRING,
  `process_time` as proctime()
) WITH (
  'connector' = 'kafka',
  ...
);

create TEMPORARY table dim_city(
  `city` STRING,
  `level` INT ,
  `province` STRING,
  `country` STRING
) WITH (
  'connector' = 'doris',
  'fenodes' = 'selectdb-cn-*******.selectdbfe.rds.aliyuncs.com:8080',
  'jdbc-url' = 'jdbc:mysql://selectdb-cn-***.selectdbfe.rds.aliyuncs.com:9030',
  'table.identifier' = 'dim.dim_city',
  'username' = 'admin',
  'password' = '****'
);

SELECT a.id, a.name, a.city, c.province, c.country,c.level 
FROM fact_table a
LEFT JOIN dim_city FOR SYSTEM_TIME AS OF a.process_time AS c
ON a.city = c.city

文法結構

source:
   type: xxx

sink:
   type: doris
   name: Doris Sink
   fenodes: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080
   username: root
   password: ""

配置項

sink.properties.column_separator: ',', -- 使用逗號分隔
-- 如果資料中可能包含逗號，建議使用不可見字元，如：
-- sink.properties.column_separator: '\x01'

'sink.properties.format' = 'json',
'sink.properties.read_json_by_line' = 'true' -- 或使用 strip_outer_array

類型映射

程式碼範例

以下為常見資料攝入情境的配置樣本。樣本中的串連地址、帳號和密碼均為預留位置，使用前需替換為實際值。

${secret_values.變數名}引用工作空間中已建立的密鑰變數，建立方式請參見變數管理。源端參數的完整說明請參見MySQL 資料攝入YAML連接器和Postgres CDC連接器。

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

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

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

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"
    # 聯合主鍵需與目標表的 Key 列一致
    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'"
    # 僅寫入以下四列
    projection: order_id, customer_id, total_amount, created_at
    primary-keys: order_id
    description: 過濾訂單狀態並裁剪列

pipeline:
  name: MySQL to SelectDB Pipeline

source:
  type: postgres
  name: PostgreSQL Source
  hostname: <yourHostname>
  port: 5432
  username: ${secret_values.pg_username}
  password: ${secret_values.pg_password}
  # PostgreSQL 的表名格式為 資料庫.Schema.表名
  tables: test_db.public.\.*
  slot.name: <yourSlotName>
  decoding.plugin.name: pgoutput
  scan.startup.mode: initial
  #（可選）使用已建立的 Publication，作業啟動時不再自動建立
  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