@@ -536,7 +536,7 @@ See more about how to use the CDC formats in [debezium-json]({% link dev/table/c
### Sink Partitioning
The config option `sink.partitioner` specifies output partitioning from Flink's partitions into Kafka's partitions.
By default, Flink uses the [Kafka default partitioner](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java) to parititon records. It uses the [sticky partition strategy](https://www.confluent.io/blog/apache-kafka-producer-improvements-sticky-partitioner/) for records with null keys and uses a murmur2 hash to compute the partition for a record with the key defined.
By default, Flink uses the [Kafka default partitioner](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java) to partition records. It uses the [sticky partition strategy](https://www.confluent.io/blog/apache-kafka-producer-improvements-sticky-partitioner/) for records with null keys and uses a murmur2 hash to compute the partition for a record with the key defined.
In order to control the routing of rows into partitions, a custom sink partitioner can be provided. The 'fixed' partitioner will write the records in the same Flink partition into the same Kafka partition, which could reduce the cost of the network connections.
<td>Specify what connector to use, for Kafka use: <code>'kafka'</code>.</td>
<td>使用的连接器，Kafka 连接器使用 <code>'kafka'</code>。</td>
</tr>
<tr>
<td><h5>topic</h5></td>
<td>required for sink, optional for source(use 'topic-pattern' instead if not set)</td>
<tdstyle="word-wrap: break-word;">(none)</td>
<td>sink 必需</td>
<tdstyle="word-wrap: break-word;">(无)</td>
<td>String</td>
<td>Topic name(s) to read data from when the table is used as source. It also supports topic list for source by separating topic by semicolon like <code>'topic-1;topic-2'</code>. Note, only one of "topic-pattern" and "topic" can be specified for sources. When the table is used as sink, the topic name is the topic to write data to. Note topic list is not supported for sinks.</td>
<td>The regular expression for a pattern of topic names to read from. All topics with names that match the specified regular expression will be subscribed by the consumer when the job starts running. Note, only one of "topic-pattern" and "topic" can be specified for sources.</td>
This can set and pass arbitrary Kafka configurations. Suffix names must match the configuration key defined in <ahref="https://kafka.apache.org/documentation/#configuration">Kafka Configuration documentation</a>. Flink will remove the "properties." key prefix and pass the transformed key and values to the underlying KafkaClient. For example, you can disable automatic topic creation via <code>'properties.allow.auto.create.topics' = 'false'</code>. But there are some configurations that do not support to set, because Flink will override them, e.g. <code>'key.deserializer'</code> and <code>'value.deserializer'</code>.
<td>Startup mode for Kafka consumer, valid values are <code>'earliest-offset'</code>, <code>'latest-offset'</code>, <code>'group-offsets'</code>, <code>'timestamp'</code> and <code>'specific-offsets'</code>.
See the following <ahref="#start-reading-position">Start Reading Position</a> for more details.</td>
<td>Specify offsets for each partition in case of <code>'specific-offsets'</code> startup mode, e.g. <code>'partition:0,offset:42;partition:1,offset:300'</code>.
<li><code>default</code>: use the kafka default partitioner to partition records.</li>
<li><code>fixed</code>: each Flink partition ends up in at most one Kafka partition.</li>
<li><code>round-robin</code>: a Flink partition is distributed to Kafka partitions sticky round-robin. It only works when record's keys are not specified.</li>
<li>Custom <code>FlinkKafkaPartitioner</code> subclass: e.g. <code>'org.mycompany.MyPartitioner'</code>.</li>
<td>Defines the delivery semantic for the Kafka sink. Valid enumerationns are <code>'at-least-once'</code>, <code>'exactly-once'</code> and <code>'none'</code>.
See <ahref='#consistency-guarantees'>Consistency guarantees</a> for more details. </td>
<td>Defines the parallelism of the Kafka sink operator. By default, the parallelism is determined by the framework using the same parallelism of the upstream chained operator.</td>
The following example shows a key and value format that both contain a `version` field:
以下示例展示了在消息键和消息体中同时包含 `version` 字段的情况：
<divclass="codetabs"markdown="1">
<divdata-lang="SQL"markdown="1">
...
...
@@ -487,93 +469,87 @@ CREATE TABLE KafkaTable (
</div>
</div>
The value format must be configured in `'EXCEPT_KEY'` mode. The formats will be configured with
the following data types:
消息体格式必须配置为 `'EXCEPT_KEY'` 模式。格式将被配置为以下的数据类型：
{% highlight text %}
key format:
消息键格式：
ROW<`version`INT,`user_id`BIGINT,`item_id`BIGINT>
value format:
消息体格式：
ROW<`version`INT,`behavior`STRING>
{% endhighlight %}
### Topic and Partition Discovery
### Topic 和 Partition 的探测
The config option `topic` and `topic-pattern` specifies the topics or topic pattern to consume for source. The config option `topic` can accept topic list using semicolon separator like 'topic-1;topic-2'.
The config option `topic-pattern` will use regular expression to discover the matched topic. For example, if the `topic-pattern` is `test-topic-[0-9]`, then all topics with names that match the specified regular expression (starting with `test-topic-` and ending with a single digit)) will be subscribed by the consumer when the job starts running.
To allow the consumer to discover dynamically created topics after the job started running, set a non-negative value for `scan.topic-partition-discovery.interval`. This allows the consumer to discover partitions of new topics with names that also match the specified pattern.
Please refer to [Kafka DataStream Connector documentation]({% link dev/connectors/kafka.zh.md %}#kafka-consumers-topic-and-partition-discovery) for more about topic and partition discovery.
If `timestamp` is specified, another config option `scan.startup.timestamp-millis` is required to specify a specific startup timestamp in milliseconds since January 1, 1970 00:00:00.000 GMT.
If `specific-offsets` is specified, another config option `scan.startup.specific-offsets` is required to specify specific startup offsets for each partition,
e.g. an option value `partition:0,offset:42;partition:1,offset:300` indicates offset `42` for partition `0` and offset `300` for partition `1`.
Flink natively supports Kafka as a changelog source. If messages in Kafka topic is change event captured from other databases using CDC tools, then you can use a CDC format to interpret messages as INSERT/UPDATE/DELETE messages into Flink SQL system.
Flink provides two CDC formats [debezium-json]({% link dev/table/connectors/formats/debezium.zh.md %}) and [canal-json]({% link dev/table/connectors/formats/canal.zh.md %}) to interpret change events captured by [Debezium](https://debezium.io/) and [Canal](https://github.com/alibaba/canal/wiki).
The changelog source is a very useful feature in many cases, such as synchronizing incremental data from databases to other systems, auditing logs, materialized views on databases, temporal join changing history of a database table and so on.
See more about how to use the CDC formats in [debezium-json]({% link dev/table/connectors/formats/debezium.zh.md %}) and [canal-json]({% link dev/table/connectors/formats/canal.zh.md %}).
请参阅 [debezium-json]({% link dev/table/connectors/formats/debezium.zh.md %}) 和 [canal-json]({% link dev/table/connectors/formats/canal.zh.md %}) 以了解如何使用 CDC 格式。
### Sink Partitioning
### Sink 分区
The config option `sink.partitioner` specifies output partitioning from Flink's partitions into Kafka's partitions.
By default, Flink uses the [Kafka default partitioner](https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java) to parititon records. It uses the [sticky partition strategy](https://www.confluent.io/blog/apache-kafka-producer-improvements-sticky-partitioner/) for records with null keys and uses a murmur2 hash to compute the partition for a record with the key defined.
In order to control the routing of rows into partitions, a custom sink partitioner can be provided. The 'fixed' partitioner will write the records in the same Flink partition into the same partition, which could reduce the cost of the network connections.
By default, a Kafka sink ingests data with at-least-once guarantees into a Kafka topic if the query is executed with [checkpointing enabled]({% link dev/stream/state/checkpointing.zh.md %}#enabling-and-configuring-checkpointing).
Please refer to [Kafka documentation]({% link dev/connectors/kafka.zh.md %}#kafka-producers-and-fault-tolerance) for more caveats about delivery guarantees.
请参阅 [Kafka 文档]({% link dev/connectors/kafka.zh.md %}#kafka-producers-and-fault-tolerance) 以获取更多关于语义保证的信息。
### Source Per-Partition Watermarks
### Source 按分区 Watermark
Flink supports to emit per-partition watermarks for Kafka. Watermarks are generated inside the Kafka
consumer. The per-partition watermarks are merged in the same way as watermarks are merged during streaming
shuffles. The output watermark of the source is determined by the minimum watermark among the partitions
it reads. If some partitions in the topics are idle, the watermark generator will not advance. You can
alleviate this problem by setting the [`'table.exec.source.idle-timeout'`]({% link dev/table/config.zh.md %}#table-exec-source-idle-timeout)
您可以通过在表配置中设置 [`'table.exec.source.idle-timeout'`]({% link dev/table/config.md %}#table-exec-source-idle-timeout) 选项来避免
上述问题。
Please refer to [Kafka watermark strategies]({% link dev/event_timestamps_watermarks.zh.md %}#watermark-strategies-and-the-kafka-connector)
for more details.
请参阅 [Kafka watermark 策略]({% link dev/event_timestamps_watermarks.md %}#watermark-strategies-and-the-kafka-connector) 一节以获取更多细节。
Data Type Mapping
数据类型映射
----------------
Kafka stores message keys and values as bytes, so Kafka doesn't have schema or data types. The Kafka messages are deserialized and serialized by formats, e.g. csv, json, avro.
Thus, the data type mapping is determined by specific formats. Please refer to [Formats]({% link dev/table/connectors/formats/index.zh.md %}) pages for more details.
