Kudu 连接器

Flink-1.12/FLINK-1.14

简述

Easystream 支持输出到 Kudu，包括upsert、insert。支持CDC

示例

• ddl

create table kuduTable(
  v_string VARCHAR,
  v_boolean BOOLEAN,
  v_byte TINYINT,
  v_short SMALLINT,
  v_int int,
  v_long BIGINT,
  v_float FLOAT,
  v_double DOUBLE,
  v_decimal DECIMAL(38,18),
  v_unixTime TIMESTAMP(3),
  v_unixTime1 TIMESTAMP(9) with local time zone,
  v_binary VARBINARY
) with(
    'connector' = 'kudu',
    'update-mode' = 'upsert',
    'masters' = 'innosql1.dg.163.org:7051,innosql2.dg.163.org:7051,innosql3.dg.163.org:7051',
    'table.name'='sloth-kudu-sink-test',
    'flush.mode'='autoFlushSync',
    'primary.keys'='v_int'
    ,'create.table.auto'='true'
    ,'ignore.delete'='true'
    ,'ignore.delete.not.exist.error'='true'

);

• 元数据方式

SET 'a.primary.keys' = 'v_string';
SET 'a.update-mode' = 'upsert';

INSERT INTO sloth_kudu_yxx.sloth_kudu_db.a
 (v_string, v_long )
 SELECT v2.v_string as v_string, cast(v2.v_long as bigint) as v_long FROM v2;

With 参数

Flink-1.10

简述

Easystream 支持输出到 Kudu。

示例

• DDL方式

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_decimal DECIMAL(38,18),
  test_double DOUBLE,
  test_timestamp TIMESTAMP(3),
  test_varbinary VARBINARY
) with (
  'connector.type'='kudu',
  'masters'='kudu1.lt.163.org:7051,kudu2.lt.163.org:7051,kudu3.lt.163.org:7051',
  'update-mode'='append',
  'primary.keys'='test_varchar',
  'flush.mode'='autoFlushSync',
  'connector.property-version'='1',
  'table.name'='kudu_table_name'
);

• 元数据方式

目标表不需要在数仓定义表 schema 信息，只需定义数据源即可

insert into [catalog].[database].[table] select * from source;

-- music_useraction_queue_3表的配置信息
set 'music_useraction_queue_3.connections.group.id' = 'test_group_id';
set 'music_useraction_queue_3.connections.startup-mode' = 'latest-offset';
-- kudu表的配置信息
SET 'sloth_kudu_ua.primary.keys' = 'v_string';
SET 'sloth_kudu_ua.update-mode' = 'upsert';
CREATE VIEW v2 AS
    SELECT os, userid, appver,logtime
    from music_useraction_queue_3;
-- 因为kudu没有database的概念，所以在表名中区分database。kudu表名规范：impala::[database].[tableName]
INSERT INTO sloth_kudu_yxx.sloth_kudu_db.sloth_kudu_ua
 (v_string, v_long )
 SELECT v2.userid as v_string, cast(v2.userid as bigint) as v_long FROM v2;

With 参数