Custom Kafka partitioner-Realtime Compute for Apache Flink(Flink)-阿里云帮助中心

Partitioning modes

The Kafka connector allows you to configure a partition strategy by setting the sink.partitioner parameter. If the built-in strategies do not meet your requirements, you can implement a custom partitioner to control record distribution.

Mode	Partitioning logic
default (default value)	Keyed records: Records with the same key are routed to the same partition, guaranteeing in-partition ordering. Keyless records: Records are distributed evenly across partitions in a round-robin fashion for load balancing.
fixed	Each parallel sink task writes to a fixed partition. If a topic has 3 partitions and the sink parallelism is 3, each parallel task writes to one dedicated Kafka partition. If a topic has 4 partitions and the sink parallelism is 3, three partitions will receive data, while one partition will remain idle. If a topic has 2 partitions and the sink parallelism is 3, one parallel task will fail because it cannot be assigned a partition.
round-robin	Records from Flink's parallel sink tasks are distributed to Kafka partitions in a round-robin fashion.

Custom partitioner for SQL jobs

Step 1: Write a custom partitioner

Your partitioner must extend the FlinkKafkaPartitioner class and override the partition method.

KafkaSinkPartitioner.java

Custom partitioning logic: This example extracts the last two digits of a date string from the record and applies a modulo 3 operation. This logic routes records from the same day to the same partition, distributing them across three partitions.

package com.aliyun;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
import org.apache.flink.table.data.GenericRowData;
public class KafkaSinkPartitioner extends FlinkKafkaPartitioner {
    @Override
    public int partition(Object record, byte[] key, byte[] value, String topic, int[] partitionSize) {
    // In Flink SQL, the record's actual type is GenericRowData.
        if (record instanceof GenericRowData){
            GenericRowData grData  = (GenericRowData) record;
            // You can use the following methods to retrieve all fields in the row.
            for (int i = 0; i < grData.getArity(); i++){
                Object field = grData.getField(i);
                System.out.println("index: " + i + " :" + field);
            }
            // Partition based on the date: use the last two digits of the date string to ensure records from the same day go to the same partition.
            String dateString = grData.getString(2).toString();
            int date = Integer.parseInt(dateString.substring(dateString.length() - 2));
            return date % 3;
        }
        else {
            throw new IllegalArgumentException("record is not GenericRowData");
        }
    }
}

Step 2: Write the SQL job

To use your custom partitioner, set the sink.partitioner parameter in your SQL job. This example uses ApsaraMQ for Kafka.

CREATE TEMPORARY TABLE KafkaSource (
  order_id STRING,
  order_name STRING,
  dt STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'source',
  'properties.group.id' = 'test-group',  -- consumer group ID
  'properties.bootstrap.servers' = '<bootstrap.servers>', -- Enter the broker endpoint.
  'format' = 'csv',                        -- Data format for the value part.
  'scan.startup.mode' = 'earliest-offset'  -- Read from the earliest offset.
);
CREATE TEMPORARY TABLE kafkaSink (
  order_id STRING,
  order_name STRING,
  dt STRING
) WITH (
  'connector' = 'kafka',
  'topic' = 'sink',
  'properties.bootstrap.servers' ='<bootstrap.servers>',
  'format' = 'csv',
  'properties.enable.idempotence' = 'false',
  'sink.partitioner' = 'com.aliyun.KafkaSinkPartitioner'
);
INSERT INTO kafkaSink
SELECT * FROM KafkaSource
;

pom.xml

Other versions of the VVR Kafka connector are available in the Maven Central Repository.

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.aliyun</groupId>
    <artifactId>KafkaPartitionerSQL</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
    <!--Kafka connector-->
        <dependency>
            <groupId>com.alibaba.ververica</groupId>
            <artifactId>ververica-connector-kafka</artifactId>
            <version>1.17-vvr-8.0.11-1</version>
        </dependency>
    <!--flink dependency-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>1.17.2</version>
        </dependency>
    </dependencies>
    <build>
        <finalName>KafkaPartitionerSQL</finalName>
        <plugins>
            <!-- Java compiler-->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.13.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
            <!-- We use maven-shade to create a fat JAR that contains all necessary dependencies. -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.5.3</version>
            </plugin>
        </plugins>
    </build>
</project>

Important

ApsaraMQ for Kafka does not support idempotent writes or transactional writes, so it cannot provide exactly-once semantics. You must add properties.enable.idempotence=false to the sink table to disable idempotent writes.
Set the sink.partitioner parameter to the fully qualified class name of your class that extends FlinkKafkaPartitioner, such as com.aliyun.KafkaSinkPartitioner.
For more information about WITH parameters, see Kafka Connector.

Step 3: Package and upload the partitioner

Use the Artifacts feature to upload the compiled JAR package to the console.

On the Artifacts tab, click Upload Artifact to upload the JAR package. Once uploaded, the file appears in the artifact list.

Step 4: Reference the JAR package

After you add the JAR package as an additional dependency, set the sink.partitioner parameter in the WITH clause to the fully qualified class name of the partitioner, such as org.mycompany.MyPartitioner.

Click More configurations, set Engine version to vvr-8.0.11-flink-1.17, and confirm that KafkaPartitionerSQL.jar is listed as an additional dependency. In the SQL editor, update the WITH clause by setting properties.bootstrap.servers to your Kafka instance's broker endpoint and sink.partitioner to the fully qualified class name of your custom partitioner, such as com.aliyun.KafkaSinkPartitioner. Finally, run the INSERT INTO kafkaSink SELECT * FROM KafkaSource; statement.

Step 5: Verify the results

Use ApsaraMQ for Kafka to test the setup by sending and receiving messages.

Log in to the ApsaraMQ for Kafka console.
On the Overview page, in the Resource Distribution section, select a region.
On the Instances page, click the name of the target instance.

On the Topics page, select the source topic and click Send Message to send test messages.

yun001,flink,20250501
yun002,flink,20250505
yun003,flink,20250505
yun004,flink,20250505

After sending the messages, check the downstream topic to see which partitions received them.

You will see that records with the same date have been written to the same partition.

On the topic details page, click the Partition Status tab and check the number of records in each partition. The distribution should be: Partition 0: 0 records, Partition 1: 1 record, and Partition 2: 3 records.

Custom partitioner for JAR jobs

Step 1: Write the JAR job

Use the setPartitioner method in the VVR Kafka connector to specify the custom partitioner class.

KafkaPartitionerDataStream.java

Custom partitioning logic: This example extracts the last two digits of a date string from the record and applies a modulo 3 operation. This logic routes records from the same day to the same partition, distributing them across three partitions.

package com.aliyun;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema;
import org.apache.flink.kafka.shaded.org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.flink.kafka.shaded.org.apache.kafka.common.serialization.StringSerializer;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.partitioner.FlinkKafkaPartitioner;
public class KafkaPartitionerDataStream {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // Build a Kafka source.
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("<BootstrapServers>")
                .setTopics("source")
                .setGroupId("test-group")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class))
                .build();
        // Use a no-watermark strategy.
        DataStreamSource<String> source = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka Source");
        // Build a Kafka sink.
        KafkaSink<String> sink = KafkaSink.<String>builder()
                .setBootstrapServers("<BootstrapServers>")
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.builder()
                                .setTopic("sink")
                                // Custom partitioner
                                .setPartitioner(new KafkaSinkPartitioner())      
                                .setKafkaValueSerializer(StringSerializer.class) 
                                .build())
                .build();
        source.sinkTo(sink).name("Kafka Sink");
        env.execute();
    }
    static class KafkaSinkPartitioner extends FlinkKafkaPartitioner<String> {
    // Partition by date: Use the last two digits of the date string to route same-day records to the same partition.
        @Override
        public int partition(String  record, byte[] key, byte[] value, String topic, int[] partitionSize) {
            String[] s = record.split(",");
            int date = Integer.parseInt(s[2].substring(s[2].length() - 2));
            return date % 3;
        }
    }
}

pom.xml

Other versions of the VVR Kafka connector are available in the Maven Central Repository.

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.aliyun</groupId>
    <artifactId>KafkaPartitionerDataStream</artifactId>
    <version>1.0-SNAPSHOT</version>
    <dependencies>
        <!--VVR Kafka Connector-->
        <dependency>
            <groupId>com.alibaba.ververica</groupId>
            <artifactId>ververica-connector-kafka</artifactId>
            <version>1.17-vvr-8.0.11-1</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-runtime</artifactId>
            <version>1.17.2</version>
            <scope>provided</scope>
        </dependency>
    </dependencies>
    <build>
        <finalName>KafkaPartitionerDS</finalName>
        <plugins>
            <!-- Java compiler-->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.13.0</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
            <!-- We use maven-shade to create a fat JAR that contains all necessary dependencies. -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.5.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.aliyun.KafkaPartitionerDataStream</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

Step 2: Package and upload the job

Use the Artifacts feature to upload the compiled JAR package to the console. We recommend setting the connector dependency scope to provided and adding it as an additional dependency. This approach simplifies version management and reduces the JAR package size for easier uploads. For more information, see Connector Dependencies and Usage.

On the Artifacts tab, click Upload Artifact to upload the JAR package to the console.

Step 3: Deploy the job

For more information about deployment parameters, see Deploy a JAR job.

In the left-side navigation pane, click Job Operations and then click Deploy. In the Deploy JAR job dialog box, configure the deployment. Set Deployment mode to stream mode, enter KafkaPartitionerDataStream for Deployment Name, select vvr-8.0.11-flink-1.17 for Engine version, select the uploaded KafkaPartitionerDS.jar for JAR URI, enter com.aliyun.KafkaPartitionerDataStream for Entry Point Class, and select ververica-connector-kafka-1.17-vvr-8.0.11-1.jar as an additional dependency. Finally, click Deploy.

Step 4: Verify the results

Use ApsaraMQ for Kafka to test the setup by sending and receiving messages.

Log in to the ApsaraMQ for Kafka console.
On the Overview page, in the Resource Distribution section, select a region.
On the Instances page, click the name of the target instance.

On the Topics page, select the source topic and click Send Message to send test messages.

yun001,flink,20250501
yun002,flink,20250505
yun003,flink,20250505
yun004,flink,20250505

After sending the messages, check the downstream topic to see which partitions received them.

You will see that records with the same date have been written to the same partition.

Custom partitioned writes with a partitioner

Partitioning modes

Custom partitioner for SQL jobs

Step 1: Write a custom partitioner

KafkaSinkPartitioner.java

Step 2: Write the SQL job

pom.xml

Step 3: Package and upload the partitioner

Step 4: Reference the JAR package

Step 5: Verify the results

Custom partitioner for JAR jobs

Step 1: Write the JAR job

KafkaPartitionerDataStream.java

pom.xml

Step 2: Package and upload the job

Step 3: Deploy the job

Step 4: Verify the results

Related documents