Kafka Connectors

Flink-1.12/FLINK-1.14

1.12简述

• kafka输入输出，支持upsert方式

1.12示例

官方文档

• ddl

CREATE TABLE user_log (
user_id VARCHAR,
item_id VARCHAR,
category_id VARCHAR,
behavior VARCHAR,
-- ts AS PROCTIME(),
ts BIGINT,
ts0 AS TO_TIMESTAMP(FROM_UNIXTIME(ts, 'yyyy-MM-dd HH:mm:ss')),
WATERMARK FOR ts0 AS ts0 - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'userlog-test',
'scan.startup.mode' = 'earliest-offset',
'properties.bootstrap.servers' = 'sloth-test2.dg.163.org:9092',
'format' = 'json'
);

CREATE TABLE usertest_sink (
  id VARCHAR,
  name varchar
) WITH (
  'connector' = 'kafka',
  'topic' = 'usertest_sink',
  'properties.bootstrap.servers' = 'sloth-test2.dg.163.org:9092',
  'properties.group.id' = 'test',
  'table.append-only' = 'true',
  'sink.partitioner' = 'round-robin',
  'sink.parallelism' = '1',
  'format' = 'json'
);

insert into usertest_sink
select user_id id, '1' name from user_log group by user_id, hop(ts0, interval '10' SECOND, interval '10' SECOND)
;

如果kafka 集群开启了sasl 方式的认证 增加如下参数

 'properties.security.protocol' = 'SASL_PLAINTEXT',
 'properties.sasl.mechanism' = 'SCRAM-HSA-256',
 'properties.sasl.jaas.config'='org.apache.kafka.common.security.plain.PlainLoginModule required username="user" password="password";'

• 1.12元数据方式

set 'user_log.scan.startup.mode' = 'earliest-offset';
set 'user_log.properties.group.id' = 'test';

insert into ztes.user_test_sink
select user_id id, '1' name from ztes.user_log
;

1.12参数

Flink-1.10

简述

• kafka输入输出，表字段支持json消息格式

示例：

• flink kafka connector官方文档 https://ci.apache.org/projects/flink/flink-docs-release-1.10/dev/table/connect.html#kafka-connector

• Time、Date、Timestamp类型

  • For example: 2018-01-01 for date, 20:43:59Z for time, and 2018-01-01T20:43:59Z for timestamp.

• source

CREATE TABLE user_log (
  user_id BIGINT,
  item_id VARCHAR,
  category_id VARCHAR,
  behavior VARCHAR,
  ts TIMESTAMP
) WITH (
  'connector.type' = 'kafka',
  'connector.version' = 'universal',
  'connector.topic' = 'xxx',
  'connector.startup-mode' = 'earliest-offset',
  'connector.properties.zookeeper.connect' = 'xxx',
  'connector.properties.bootstrap.servers' = 'xxx',
  'connector.properties.group.id' = 'xxx',
  'update-mode' = 'append',
  'format.type' = 'json',
  'format.derive-schema' = 'true',
  'format.fail-on-missing-field' = 'false'
);

• sink

  CREATE TABLE user_age (
  user_id BIGINT,
  item_id VARCHAR,
  age INT,
  behavior VARCHAR,
  ts TIMESTAMP
  ) WITH (
  'connector.type' = 'kafka',
  'connector.version' = 'universal',
  'connector.topic' = 'xxx',
  'connector.properties.zookeeper.connect' = 'xxx',
  'connector.properties.bootstrap.servers' = 'xxx',
  'connector.properties.group.id' = 'xxx',
  'update-mode' = 'append',
  'connector.sink-partitioner' = 'round-robin',
  'format.type' = 'json',
  'format.derive-schema' = 'true'
  );

  with参数

source参数

• connector.type
  • 'kafka' 必选。
• connector.version
  • 某kafka版本；必选。universal。版本对应关系见Kafka版本对应关系。
• connector.topic
  • kafka的topic；必选
• connector.startup-mode
  • 启动方式，可选。earliest-offset、latest-offset、group-offsets和specific-offsets，当指定为specific-offsets时，还需要配置如下：

    'connector.specific-offsets.0.partition' = '0',
    'connector.specific-offsets.0.offset' = '42',
    'connector.specific-offsets.1.partition' = '1',
    'connector.specific-offsets.1.offset' = '300'

• connector.properties.*.key
  • 配置kafka的property key，必选。
• connector.properties.*.value
  • 配置kafka的property value，必选。

    'connector.properties.zookeeper.connect' = 'xxx',
    'connector.properties.bootstrap.servers' = 'xxx',
    'connector.properties.group.id' = 'xxx',
    或者是
    'connector.properties.0.key' = 'zookeeper.connect', -- optional: connector specific properties
    'connector.properties.0.value' = 'localhost:2181',
    'connector.properties.1.key' = 'bootstrap.servers',
    'connector.properties.1.value' = 'localhost:9092',
    'connector.properties.2.key' = 'group.id',
    'connector.properties.2.value' = 'testGroup',

• flink.partition-discovery.interval-millis
  • 定时检查是否有新分区产生；可选；默认不启用，单位为毫秒。

sink 参数

kafka版本对应配置