FlinkCDC（Change Data Capture）Kafka 是一個用于捕獲和跟蹤 Kafka 集群中數據變更的 Flink 連接器。在使用 FlinkCDC Kafka 時，可以通過配置分區策略來控制如何將變更數據分布到不同的 Kafka 分區中。以下是一些常見的分區策略：
基于 key 的哈希分區： 在這種策略中，Flink 會根據變更數據的 key 計算哈希值，并將其映射到 Kafka 分區。這樣可以確保具有相同 key 的變更數據始終發送到同一個分區。這種策略適用于需要保證相同 key 的變更數據順序一致性的場景。
配置示例：
Properties kafkaProperties = new Properties();
kafkaProperties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
kafkaProperties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
kafkaProperties.setProperty("group.id", "flink_cdc_consumer");
kafkaProperties.setProperty("enable.auto.commit", "false");
kafkaProperties.setProperty("auto.offset.reset", "earliest");
kafkaProperties.setProperty("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RoundRobinAssignor");
基于 key 的模分區： 在這種策略中，Flink 會根據變更數據的 key 計算模值，并將其映射到 Kafka 分區。這樣可以確保具有相同 key 的變更數據始終發送到同一個分區。這種策略適用于需要保證相同 key 的變更數據順序一致性的場景。
配置示例：
Properties kafkaProperties = new Properties();
kafkaProperties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
kafkaProperties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
kafkaProperties.setProperty("group.id", "flink_cdc_consumer");
kafkaProperties.setProperty("enable.auto.commit", "false");
kafkaProperties.setProperty("auto.offset.reset", "earliest");
kafkaProperties.setProperty("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RoundRobinAssignor");
kafkaProperties.setProperty("properties.key.partitioner.class", "org.apache.kafka.clients.producer.internals.DefaultPartitioner");
kafkaProperties.setProperty("properties.key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
基于 value 的哈希分區： 在這種策略中，Flink 會根據變更數據的 value 計算哈希值，并將其映射到 Kafka 分區。這樣可以確保具有相同 value 的變更數據始終發送到同一個分區。這種策略適用于需要保證相同 value 的變更數據順序一致性的場景。
配置示例：
Properties kafkaProperties = new Properties();
kafkaProperties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
kafkaProperties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
kafkaProperties.setProperty("group.id", "flink_cdc_consumer");
kafkaProperties.setProperty("enable.auto.commit", "false");
kafkaProperties.setProperty("auto.offset.reset", "earliest");
kafkaProperties.setProperty("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RoundRobinAssignor");
kafkaProperties.setProperty("properties.value.partitioner.class", "org.apache.kafka.clients.producer.internals.DefaultPartitioner");
kafkaProperties.setProperty("properties.value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
基于 value 的模分區： 在這種策略中，Flink 會根據變更數據的 value 計算模值，并將其映射到 Kafka 分區。這樣可以確保具有相同 value 的變更數據始終發送到同一個分區。這種策略適用于需要保證相同 value 的變更數據順序一致性的場景。
配置示例：
Properties kafkaProperties = new Properties();
kafkaProperties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
kafkaProperties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
kafkaProperties.setProperty("group.id", "flink_cdc_consumer");
kafkaProperties.setProperty("enable.auto.commit", "false");
kafkaProperties.setProperty("auto.offset.reset", "earliest");
kafkaProperties.setProperty("partition.assignment.strategy", "org.apache.kafka.clients.consumer.RoundRobinAssignor");
kafkaProperties.setProperty("properties.value.partitioner.class", "org.apache.kafka.clients.producer.internals.DefaultPartitioner");
kafkaProperties.setProperty("properties.value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
在選擇分區策略時，需要根據具體的應用場景和需求來決定。例如，如果需要保證相同 key 的變更數據順序一致性，可以選擇基于 key 的哈希分區或模分區策略。如果需要保證相同 value 的變更數據順序一致性，可以選擇基于 value 的哈希分區或模分區策略。
