实现一个 kafka 连接自定义分区器答案

【问题标题】：Implementing a kafka connect custom partitioner实现一个 kafka 连接自定义分区器
【发布时间】：2019-12-21 06:45:05
【问题描述】：

我正在使用 confluent 的 kafka 连接将数据传输到 s3 存储桶中。理想情况下基于密钥进行分区。由于现有的 FieldPartitioner 仅适用于 Avro 模式记录，而不适用于一般的字符串化 JSON 文本。我想我会写自己的连接器。

课程如下：

package com.package.kafka.connect;


import com.google.gson.JsonElement;
import com.google.gson.JsonParser;

import io.confluent.connect.storage.partitioner.DefaultPartitioner;
import io.confluent.connect.storage.partitioner.FieldPartitioner;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.connect.sink.SinkRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;

public class JsonFieldPartitioner<T> extends DefaultPartitioner<T> {

    private static final Logger log = LoggerFactory.getLogger(FieldPartitioner.class);
    private List<String> fieldNames;
    private List<String> keys;

    public void configure(Map<String, Object> config){
        fieldNames = (List<String>) config.get("partition.field.name");
        String field =  fieldNames.get(0);
        keys = new ArrayList<String>(Arrays.asList(field.split(".")));
    }

    public String encodePartition(SinkRecord sinkRecord){
        String value = sinkRecord.value().toString();

        JsonElement rootElement = new JsonParser().parse(value);
        JsonElement element = rootElement;

        for(String key : keys){
            log.info("key: "+ key);
            try{
                element = element.getAsJsonObject().get(key);
            }catch(Exception e){
                log.error("encountered error getting key: " + key);
                throw new ConfigException("Key element not found" + key);
            }
        }

        String fieldValue  = "";
        try{
            fieldValue =  element.getAsString();
        }catch(Exception e){
            log.error("encountered error getting key value ");
            throw new ConfigException("Key element not found");
        }
        return fieldValue;
    }

    public List<T> partitionFields() {
        if (partitionFields == null) {
            partitionFields = newSchemaGenerator(config).newPartitionFields(
                    Utils.join(fieldNames, ",")
            );
        }
        return partitionFields;
    }
}

当我构建它并尝试运行 kafka connect 时出现错误

java.lang.NullPointerException
    at io.confluent.connect.storage.partitioner.PartitionerConfig.classNameEquals(PartitionerConfig.java:269)
    at io.confluent.connect.storage.partitioner.PartitionerConfig.access$000(PartitionerConfig.java:32)
    at io.confluent.connect.storage.partitioner.PartitionerConfig$PartitionerClassDependentsRecommender.visible(

来自Packaging a custom Java `partitioner.class` plugin for Kafka Connect in Confluent 4.1 + Kafka 1.1? 我尝试将由此构建的 jar 文件放在 $CONFLUENT_HOME 的 kafka-connect-storage-common 目录中，但仍然出现相同的错误。

构建jar的gradle文件在这里

    id 'java'
}

group 'JsonFieldPartitioner'
version '1.0-SNAPSHOT'

sourceCompatibility = 1.8

repositories {
    mavenCentral()
}

dependencies {
    compile group: 'org.apache.kafka', name: 'connect-api', version: '2.3.0'
    compile fileTree( dir:'/Users/myuser/confluent-5.3.0/share/java/kafka-connect-storage-common', include: ['*.jar'])
    compile group: 'joda-time', name: 'joda-time', version: '2.10.3'
    compile group: 'com.google.code.gson', name: 'gson', version: '2.8.5'
    testCompile group: 'junit', name: 'junit', version: '4.12'
}

在 s3 连接器属性文件中，我只是通过 com.package.kafka.connect.JsonFieldPartitioner 引用该类

如果有人成功构建了自定义分区器，我们将不胜感激。

【问题讨论】：

标签： java apache-kafka apache-kafka-connect

【解决方案1】：

将你的自定义jar文件复制到插件文件夹下例子 /usr/local/share/kafka/plugins/confluentinc-kafka-connect-s3/

【讨论】：

有什么方法可以让它变得简单吗？如果我们需要与接收器/源连接器共享同一个分区器怎么办？这是否意味着我们也需要修改他们的kafka-connect-storage-partitioner.jar？
你不需要修改任何东西。只需在插件文件夹中复制您的自定义 jar。示例 /usr/local/share/kafka/plugins/confluentinc-kafka-connect-s3/
是的，您的回答已被编辑。我的评论不再有意义。应对罐子是个好方法。

【解决方案2】：

在准备 kafka 连接图像时：- 我们可以这样做

confluent-hub install --no-prompt /tmp/connect-fieldandtime-partitioner-1.2.0.zip

这里的 zip 是自定义分区器，那么为什么我们需要将 JAR 放在 Service 的 lib 文件夹中

【讨论】：