Upsert Kafka结果表
功能描述
DLI将Flink作业的输出数据以upsert的模式输出到Kafka中。
Apache Kafka是一个快速、可扩展的、高吞吐、可容错的分布式发布订阅消息系统，具有高吞吐量、内置分区、支持数据副本和容错的特性，适合在大规模消息处理场景中使用。
前提条件
Kafka是线下集群，需要通过增强型跨源连接功能将Flink作业与Kafka进行对接。且用户可以根据实际所需设置相应安全组规则。
注意事项
对接的Kafka集群不支持开启SASL_SSL。
语法格式
1 2 3 4 5 6 7 8 9 10 11 12 |
create table kafkaSource( attr_name attr_type (',' attr_name attr_type)* (','PRIMARY KEY (attr_name, ...) NOT ENFORCED) ) with ( 'connector.type' = 'upsert-kafka', 'connector.version' = '', 'connector.topic' = '', 'connector.properties.bootstrap.servers' = '', 'format.type' = '' ); |
参数说明
|
参数 |
是否必选 |
说明 |
|---|---|---|
|
connector.type |
是 |
connector类型，对于upsert kafka，需配置为'upsert-kafka' |
|
connector.version |
否 |
Kafka版本，仅支持：'0.11' |
|
format.type |
是 |
数据序列化格式，支持：'csv', 'json'及'avro'等 |
|
connector.topic |
是 |
kafka topic名 |
|
connector.properties.bootstrap.servers |
是 |
kafka brokers地址，以逗号分隔 |
|
connector.sink-partitioner |
否 |
记录分区方式，支持：'fixed', 'round-robin'及'custom' |
|
connector.sink-partitioner-class |
否 |
'sink-partitioner'为'custom'时，需配置，如'org.mycompany.MyPartitioner' |
|
connector.sink.ignore-retraction |
否 |
是否忽略回撤消息，默认为false。回撤消息将以null值写入kafka |
|
update-mode |
否 |
支持：'append', 'retract'及'upsert'三种写入模式 |
|
connector.properties.* |
否 |
配置kafka任意原生属性 |
示例
create table upsertKafkaSink( car_id STRING, car_owner STRING, car_brand STRING, car_speed INT, primary key (car_id) not enforced ) with ( 'connector.type' = 'upsert-kafka', 'connector.version' = '0.11', 'connector.topic' = 'test-topic', 'connector.properties.bootstrap.servers' = 'xx.xx.xx.xx:9092', 'format.type' = 'csv' );
