Realtime Compute for Apache Flink:SelectDB

Usage

1. Click JAR package to download the SelectDB custom connector (versions 1.15 to 1.17).

2. In the Realtime Compute for Apache Flink development console, upload the SelectDB custom connector. For more information, see Manage custom connectors.

3. Use the SelectDB custom connector in an SQL job. The value of the connector parameter is fixed to doris.

Syntax structure

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 parameters

• General

  Parameter	Description	Data type	Required	Default value	Remarks
  connector	The table type.	String	Yes	None	The value is fixed to doris.
  fenodes	The access address and HTTP protocol port of the ApsaraDB for SelectDB instance.	String	Yes	None	You can get the VPC Address (or Public Address) and HTTP Protocol Port from the Instance Details > Network Information page in the ApsaraDB for SelectDB console. Example: selectdb-cn-****.selectdbfe.rds.aliyuncs.com:8080.
  jdbc-url	The Java Database Connectivity (JDBC) connection information.	String	No	None	You can obtain the VPC Address (or Public Address) and MySQL Protocol Port from the Instance Details > Network Information page in the ApsaraDB for SelectDB console. Example: jdbc:mysql://selectdb-cn-***.selectdbfe.rds.aliyuncs.com:9030.
  table.identifier	The database table name.	String	Yes	None	Example: db.tbl.
  username	The username.	String	Yes	None	If you forget the password, you can reset it in the upper-right corner of the Instance Details page in the ApsaraDB for SelectDB console.
  password	The password.	String	Yes	None
  doris.request.retries	The number of retries for sending a request.	Integer	No	3	None.
  doris.request.connect.timeout	The connection timeout for sending a request.	Duration	No	30s	None.
  doris.request.read.timeout	The read timeout for sending a request.	Duration	No	30s	None.

• Source table specific

  Parameter	Description	Data type	Required	Default value	Remarks
  doris.request.query.timeout	The query timeout. The default value is 6 hours.	Duration	No	21600s	The value is fixed to doris.
  doris.request.tablet.size	The number of tablets corresponding to a partition.	Integer	No	1	A smaller value generates more partitions. This increases the degree of parallelism on the Flink side but also puts more pressure on the database.
  doris.batch.size	The maximum number of rows to read from a BE at a time.	Integer	No	4064	Increasing this value can reduce the number of connections established between Flink and the database. This reduces the extra time overhead caused by network latency.
  doris.exec.mem.limit	The memory limit for a single query.	Integer	No	8192mb	The default value is 8 GB. The unit is bytes.
  source.use-flight-sql	Specifies whether to use Arrow Flight SQL for reading.	Boolean	No	false	No configuration is required. Simply go to the Instance Details > Network Information page in the ApsaraDB for SelectDB console and click Enable Direct Cluster Connection.
  source.flight-sql-port	The arrow_flight_sql_port of the FE when reading with Arrow Flight SQL.	Integer	No	-	None.

• Sink table specific

  Parameter	Description	Data type	Required	Default value	Remarks
  sink.label-prefix	The label prefix used for Stream Load imports.	String	No	--	The prefix must be globally unique across multiple jobs to ensure the exactly-once semantics of Flink. The same label can be imported only once to prevent duplicate writes.
  sink.properties.*	The import parameters for Stream Load.	String	No	--	CSV format configuration 'sink.properties.column_separator' = ',', -- Use a comma as the delimiter -- If the data may contain commas, use a non-printable character, such as: -- 'sink.properties.column_separator' = '\x01' JSON format configuration 'sink.properties.format' = 'json', 'sink.properties.read_json_by_line' = 'true' -- or use strip_outer_array
  sink.enable-delete	Specifies whether to enable deletion. This option requires the Doris table to have batch deletion enabled.	Boolean	No	true	Only the Unique model is supported.
  sink.enable-2pc	Specifies whether to enable the two-phase commit protocol (2PC).	Boolean	No	true	Ensures exactly-once semantics. For more information about 2PC, see Explicit Transaction Operations.
  sink.buffer-size	The size of the data write cache buffer.	Integer	No	1 MB	The unit is bytes. We recommend that you do not modify this parameter. Use the default configuration.
  sink.buffer-count	The number of data write cache buffers.	Integer	No	3	We recommend that you do not modify this parameter. Use the default configuration.
  sink.max-retries	The maximum number of retries after a commit fails.	Integer	No	3	None.
  sink.enable.batch-mode	Specifies whether to use batch mode for writing.	Boolean	No	false	If enabled, the write timing does not depend on checkpoints. It is controlled by the sink.buffer-flush.max-rows/sink.buffer-flush.max-bytes/sink.buffer-flush.interval parameters. If enabled, exactly-once semantics are not guaranteed. However, you can achieve idempotence using the Uniq model.
  sink.flush.queue-size	The size of the cache queue in batch mode.	Integer	No	2	None.
  sink.buffer-flush.max-rows	The maximum number of data rows to write in a single batch in batch mode.	Integer	No	500000	None.
  sink.buffer-flush.max-bytes	The maximum number of bytes to write in a single batch in batch mode.	Integer	No	100 MB	The unit is bytes.
  sink.buffer-flush.interval	The interval for asynchronously flushing the cache in batch mode.	String	No	10s	The unit is milliseconds.
  sink.ignore.update-before	Specifies whether to ignore update-before events.	Boolean	No	true	None.

• Dimension table specific

  Parameter	Description	Data type	Required	Default value	Remarks
  lookup.cache.max-rows	The maximum number of rows for the lookup cache.	Integer	No	-1	A value of -1 means that caching is disabled by default.
  lookup.cache.ttl	The maximum time-to-live (TTL) for the lookup cache.	String	No	10s	The unit is milliseconds.
  lookup.max-retries	The number of retries after a lookup query fails.	Integer	No	1	None.
  lookup.jdbc.async	Specifies whether to enable asynchronous lookup.	Boolean	No	false	None.
  lookup.jdbc.read.batch.size	The maximum batch size for each query in asynchronous lookup.	Integer	No	128	None.
  lookup.jdbc.read.batch.queue-size	The size of the intermediate buffer queue during asynchronous lookup.	Integer	No	256	None.
  lookup.jdbc.read.thread-size	The number of JDBC lookup threads in each task.	Integer	No	3	None.

Usage examples

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' -- The same label can be imported only once to prevent duplicate writes.
);

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

Syntax structure

source:
   type: xxx

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

Configuration items

sink.properties.column_separator: ',' # Use a comma as the delimiter
# If the data may contain commas, use a non-printable character, such as:
# sink.properties.column_separator: '\x01'

sink.properties.format: 'json'
sink.properties.read_json_by_line: 'true' # or use strip_outer_array

Type mapping