TSDB Sink

简述

Easystream 支持输出到 TSDB。

Flink-1.12

示例

--SQL
--********************************************************************--
--Author: someone
--CreateTime: 2021-11-17 11:06:01
--Comment: 请输入业务注释信息
--********************************************************************--
CREATE TABLE source (
  test_int INT,
  test_varchar VARCHAR,
  test_boolean BOOLEAN,
  test_bigint BIGINT,
  test_float FLOAT,
  --  testDecimal DECIMAL(38, 18),
  test_double DOUBLE,
  PRIMARY KEY(test_int) NOT ENFORCED
) WITH ('connector' = 'datagen');

CREATE TABLE sink (
  test_int INT,
  test_varchar VARCHAR,
  test_boolean BOOLEAN,
  test_bigint BIGINT,
  test_float FLOAT,
  --  testDecimal DECIMAL(38, 18),
  test_double DOUBLE
) WITH (
  'connector' = 'tsdb',
  -- required: specify this table type is jdbc
  'master' = 'xxxx:xx',
  -- required: JDBC DB url
  'username' = 'xxxx',
  -- required: jdbc table name
  'password' = 'xxxxxx',
  -- optional: jdbc user name and password
  'database' = 'xxxxx',
  'measurement' = 'xxxxxx',
  'tags' = 'test_varchar',
  'time-field' = 'test_bigint'
);
INSERT INTO
  sink
select
  test_int,
  test_varchar,
  test_boolean,
  1111111 as test_bigint,
  test_float,
  --  testDecimal DECIMAL(38, 18),
  test_double
FROM
  source;

With 参数

Flink-1.10

示例

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,
  `tag.tag_field1` VARCHAR,
  `time.updateTime` TIMESTAMP(3)
) with(
  'connector.type'='tsdb',
  'url' = 'sloth-tsdb0.dg.163.org:8086',
  'username' = 'flink-metrics',
  'password' = '***',
  'database' = 'test-yxx',
  'measurement' = 'testtable',
  'mode' = 'cluster',
  'batch.enabled'.' = 'true',
  'actions' = '1000',
  'buffer.limit' = '10000',
  'flush.duration' = '1000',
  'jitter.duration' = '0',
  'consistency.level' = 'one',
  'enable.gzip' = 'false',
  'create.database' = 'false',
  'retention.policy' = 'autogen'
);

TSDB key 说明

• tsdb tag key ：e.g. tag.tag_field1 使用 tag. 作为前缀表明 tag_field1 是 tag key；

• tsdb time key：默认使用 time 字段存储于 tsdb；如果自行指定 point 的时间戳，请使用 time. Timestamp，使用 time. 作为前缀，且为 Timestamp 类型

• tsdb field key：除了以上两种字段，其他都为 filed key，且必须含有一个 field key。

With 参数