Flink-1.12

简述

支持starrocks/doris的写（使用stream_load优化性能）。

starrocks/doris的读、维表Join、写均可使用jdbc connector

示例

ddl方式

-- source
CREATE TABLE kafka_source (
      test_int INT,
      test_varchar VARCHAR,
      test_boolean BOOLEAN,
      test_tinyint TINYINT,
      test_smallint SMALLINT,
      test_bigint BIGINT,
      test_float FLOAT,
      test_double DOUBLE
) WITH (
  'connector' = 'memory',
  'table.name' = 'kafka_source'
);

-- sink
CREATE TABLE sink (
      test_int INT,
      test_varchar VARCHAR,
      test_boolean BOOLEAN,
      test_tinyint TINYINT,
      test_smallint SMALLINT,
      test_bigint BIGINT,
      test_float FLOAT,
      test_double DOUBLE
) WITH (
      'connector' = 'starrocks', -- required: specify this table type is jdbc
      'jdbc-url' = 'jdbc:mysql://hadoop336.photo.163.org:9030', -- required: JDBC DB url
      'load-url'='hadoop336.photo.163.org:8030',
      'table-name' = 'allType',  -- required: jdbc table name
      'database-name'='data_transform',
      'username' = 'root', -- optional: jdbc user name and password
      'password' = '123456'
);

INSERT INTO sink select * FROM kafka_source;

元数据中心方式

CREATE TABLE kafka_source (
      test_int INT,
      test_varchar VARCHAR,
      test_boolean BOOLEAN,
      test_tinyint TINYINT,
      test_smallint SMALLINT,
      test_bigint BIGINT,
      test_float FLOAT,
--  testDecimal DECIMAL(38, 18),
      test_double DOUBLE
) WITH (
      'connector' = 'memory',
      'table.name' = 'kafka_source'
);

INSERT INTO bdms.data_transform.allType select * FROM kafka_source;

参数配置

• 注意
  1. Flush 在 at-least-once 的触发时机： cachedRows >= ${sink.buffer-flush.max-rows} || cachedBytes >= ${sink.buffer-flush.max-bytes} || idleTime >= ${sink.buffer-flush.interval-ms}
  2. sink.buffer-flush.{max-rows|max-bytes|interval-ms} 在 exactly-once 时失效.

     Metrics

Type mappings

注意：当doris 0.14表中存在tinyint(1)，读数据会报错

Sink

Flink-1.14

简述

支持starrocks/doris的读、维表（使用thrift rpc读数据）、写（使用stream_load优化性能）。

示例

ddl方式

-- source
CREATE TABLE kafka_source (
      test_int INT,
      test_varchar VARCHAR,
      test_boolean BOOLEAN,
      test_tinyint TINYINT,
      test_smallint SMALLINT,
      test_bigint BIGINT,
      test_float FLOAT,
      test_double DOUBLE
) WITH (
  'connector' = 'memory',
  'table.name' = 'kafka_source'
);

-- sink
CREATE TABLE sink (
      test_int INT,
      test_varchar VARCHAR,
      test_boolean BOOLEAN,
      test_tinyint TINYINT,
      test_smallint SMALLINT,
      test_bigint BIGINT,
      test_float FLOAT,
      test_double DOUBLE
) WITH (
      'connector' = 'starrocks', -- required: specify this table type is jdbc
      'jdbc-url' = 'jdbc:mysql://hadoop336.163.org:9030', -- required: JDBC DB url
      'load-url'='hadoop336.163.org:8030',
      'table-name' = 'allType',  -- required: jdbc table name
      'database-name'='data_transform',
      'username' = 'root', -- optional: jdbc user name and password
      'password' = '123456'
);

INSERT INTO sink select * FROM kafka_source;

// source
CREATE TABLE flink_test (
    date_1 DATE,
    datetime_1 TIMESTAMP(6),
    char_1 CHAR(20),
    varchar_1 VARCHAR,
    boolean_1 BOOLEAN,
    tinyint_1 TINYINT,
    smallint_1 SMALLINT,
    int_1 INT,
    bigint_1 BIGINT,
    largeint_1 STRING,
    float_1 FLOAT,
    double_1 DOUBLE,
    decimal_1 DECIMAL(27,9)
) WITH (
   'connector'='starrocks',
   'load-url'='fe_ip1:8030,fe_ip2:8030,fe_ip3:8030',
   'jdbc-url'='jdbc:mysql://fe_ip:9030',
   'username'='root',
   'password'='',
   'database-name'='flink_test',
   'table-name'='flink_test'
);

select date_1, smallint_1 from flink_test where char_1 <> 'A' and int_1 = -126

元数据中心方式

CREATE TABLE kafka_source (
      test_int INT,
      test_varchar VARCHAR,
      test_boolean BOOLEAN,
      test_tinyint TINYINT,
      test_smallint SMALLINT,
      test_bigint BIGINT,
      test_float FLOAT,
--  testDecimal DECIMAL(38, 18),
      test_double DOUBLE
) WITH (
      'connector' = 'memory',
      'table.name' = 'kafka_source'
);

INSERT INTO bdms.data_transform.allType select * FROM kafka_source;

参数配置

source

sink

• 注意
  1. Flush 在 at-least-once 的触发时机： cachedRows >= ${sink.buffer-flush.max-rows} || cachedBytes >= ${sink.buffer-flush.max-bytes} || idleTime >= ${sink.buffer-flush.interval-ms}
  2. sink.buffer-flush.{max-rows|max-bytes|interval-ms} 在 exactly-once 时失效.

     Metrics

Type mappings

注意：

1. 当doris 0.14表中存在tinyint(1)、boolean时，读数据会报错。详见：doris bug。此时可使用jdbc connector读。
2. 当任务失败时，exactly-once 语义无法保证.
3. 只支持不带聚合的SQL，如 select {*|columns|count(1)} from {table-name} where ....

   source

Sink