# Upsert Kafka结果表
#### 功能描述
DLI将Flink作业的输出数据以upsert的模式输出到Kafka中。
Apache Kafka是一个快速、可扩展的、高吞吐、可容错的分布式发布订阅消息系统，具有高吞吐量、内置分区、支持数据副本和容错的特性，适合在大规模消息处理场景中使用。
#### 前提条件
Kafka是线下集群，需要通过增强型跨源连接功能将Flink作业与Kafka进行对接。且用户可以根据实际所需设置相应安全组规则。
#### 注意事项
对接的Kafka集群不支持开启SASL_SSL。
#### 语法格式
```
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' = ''
);
```
#### 参数说明
表1参数说明 
| 参数                                     | 是否必选 | 说明                                                              |
|:---|:---|:---|
| 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'
);
```