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

類型映射