Realtime Compute for Apache Flink:Upsert Kafka connector

SQL

The Upsert Kafka connector can be used to read data from and write data to Kafka topics in the upsert fashion.

Syntax

CREATE TABLE upsert_kafka_sink(
user_region STRING,
pv BIGINT,
uv BIGINT,
PRIMARY KEY(user_region) NOT ENFORCED
)WITH(
'connector'='upsert-kafka',
'topic'='<yourTopicName>',
'properties.bootstrap.servers'='...',
'key.format'='avro',
'value.format'='avro'
);

Parameters in the WITH clause

• Common parameters

  Parameter	Description	Data type	Required	Default value	Remarks
  connector	The type of the table.	String	Yes	No default value	Set the value to upsert-kafka.
  properties.bootstrap.servers	The IP addresses or endpoints and port numbers of Kafka brokers.	String	Yes	No default value	Format: host:port,host:port,host:port. Separate multiple host:port pairs with commas (,).
  properties.*	The parameters that are configured for the Kafka client.	String	No	No default value	The suffix of this parameter must comply with the rules that are defined in Producer Configs and Consumer Configs. Flink removes the properties. prefix and passes the transformed keys and values to the Kafka client. For example, you can set properties.allow.auto.create.topics to false to disable automatic topic creation. You cannot modify the configurations of the following parameters by adding the properties. prefix, because the values of the parameters are overwritten after you use the Upsert Kafka connector: key.deserializer value.deserializer
  key.format	The format used to read or write the key field of Kafka messages.	String	Yes	No default value	You must configure the key.fields or key.fields-prefix parameter if you configure this parameter. Valid values: csv json avro debezium-json canal-json maxwell-json avro-confluent raw
  key.fields-prefix	A custom prefix for all key fields in Kafka messages. You can configure this parameter to prevent name conflicts with the value fields.	String	No	No default value	This parameter is used only to distinguish the column names of source tables and sink tables. The prefix is removed from the column names when the key fields of Kafka messages are parsed and generated. Note If you configure this parameter, you must set the value.fields-include parameter to EXCEPT_KEY.
  value.format	The format used to read or write the value field of Kafka messages.	String	Yes	No default value	The configuration of this parameter is equivalent to the configuration of the format parameter. The format parameter cannot be used together with the value.format parameter. If you configure both parameters, a conflict occurs.
  value.fields-include	Specifies whether to include the fields that correspond to message keys when the value fields of Kafka messages are parsed or generated.	String	Yes	ALL	Valid values: ALL: All fields are processed as value fields of Kafka messages. This is the default value. EXCEPT_KEY: All fields except for the fields specified by the key.fields parameter are processed as value fields of Kafka messages.
  topic	The name of the topic from which data is read or to which data is written.	String	Yes	No default value	N/A

• Parameters only for sink tables

  Parameter	Description	Data type	Required	Default value	Remarks
  sink.parallelism	The parallelism of operators in the Kafka sink table.	Integer	No	The parallelism of upstream operators, which is determined by the framework.	N/A
  sink.buffer-flush.max-rows	The maximum number of data records that can be cached before the cache is refreshed.	Integer	No	0 (disabled)	If the sink table receives a large number of updates on the same key, only the last data record of the key is retained in the cache. In this case, data caching in the sink table helps reduce the amount of data that is written to Kafka topics. This prevents potential tombstone messages from being sent to Kafka topics. Note If you want to enable data caching for sink tables, you must set the sink.buffer-flush.max-rows and sink.buffer-flush.interval parameters to values that are greater than 0.
  sink.buffer-flush.interval	The interval at which the cache is refreshed.	Duration	No	0 (disabled)	The unit can be milliseconds, seconds, minutes, or hours. For example, you can configure 'sink.buffer-flush.interval'='1 s'. If the sink table receives a large number of updates on the same key, only the last data record of the key is retained in the cache. In this case, data caching in the sink table helps reduce the amount of data that is written to Kafka topics. This prevents potential tombstone messages from being sent to Kafka topics. Note If you want to enable data caching for sink tables, you must set the sink.buffer-flush.max-rows and sink.buffer-flush.interval parameters to values that are greater than 0.

Data ingestion

The Upsert Kafka connector can be used to develop YAML drafts for data ingestion. In addition, it can be used as a sink for data ingestion. The data is written in the JSON format to the sink, and the primary key field is also included in the message body.

Syntax

sink:
  type: upsert-kafka
  name: upsert-kafka Sink
  properties.bootstrap.servers: localhost:9092
  # ApsaraMQ for Kafka
  aliyun.kafka.accessKeyId: ${secret_values.kafka-ak}
  aliyun.kafka.accessKeySecret: ${secret_values.kafka-sk}
  aliyun.kafka.instanceId: ${instancd-id}
  aliyun.kafka.endpoint: ${endpoint}
  aliyun.kafka.regionId: ${region-id}

Parameters