Kafka Connectors

Flink-1.12

简述

• kafka输入输出，支持upsert方式，见 upsertKafka.md

示例

官方文档

• 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认证配置

以下内容介绍kafka 的几种认证配置方式，详情请参考kafka 官方配置

配置方法

数据源注册时（Metahub） 的方式 ，新增自定义配置，增加如下3组配置，注意user ,password 需要替换为集群自己的账号密码，最后一项配置必须带个分号；

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

set方式配置 同时产品也支持在任务sql 中通过 set 参数进行设置，如下：

set  'tablename.properties.security.protocol' = 'SASL_PLAINTEXT',
set  'tablename.properties.sasl.mechanism' = 'PLAIN',
set 'tablename.properties.sasl.jaas.config'='org.apache.kafka.common.security.plain.PlainLoginModule required username="user" password="password";'

认证方式

首先需要确定kafka 集群使用了那种认证方式，不同的认证方式，需要使用不同的配置 kafka目前共有4种常见的认证方式
SASL/GSSAPI（kerberos）：kafka0.9版本推出，即借助kerberos实现用户认证

 'properties.security.protocol' = 'SASL_PLAINTEXT',(or SASL_SSL)
 'properties.sasl.mechanism' = 'GSSAPI', 
 'properties.sasl.kerberos.service.name'='kafka',
 'properties.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required \
    useKeyTab=true \
    storeKey=true  \
    keyTab="/etc/security/keytabs/kafka_client.keytab" \
    principal="kafka-client-1@EXAMPLE.COM";'

SASL/PLAIN：kafka0.10推出，非常简单，简单得有些鸡肋，不建议生产环境使用，除非对这个功能二次开发

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

SASL/SCRAM：kafka0.10推出，全名Salted Challenge Response Authentication Mechanism，为解决SASL/PLAIN的不足而生，缺点可能是某些客户端并不支持这种方式认证登陆（使用比较复杂）

 'properties.security.protocol'='SASL_SSL',
 'properties.sasl.mechanism'='SCRAM-SHA-256' (or SCRAM-SHA-512),
 'properties.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required username="alice"  password="alice-secret";'

SASL/OAUTHBEARER：kafka2.0推出，实现较为复杂，目前业内应该较少实践

 'properties.security.protocol'='SASL_SSL' (or SASL_PLAINTEXT if non-production),
 'properties.sasl.mechanism'='OAUTHBEARER',
 'properties.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required unsecuredLoginStringClaim_sub="alice";'

• 元数据方式

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
;

参数

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
```SQL
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版本对应配置