实现kafka连接自定义分区程序

Question

我使用confluent的kafka connect将数据传输到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(

从查看打包一个自定义Java'partitioner.class'插件为Kafka连接在汇流4.1 Kafka 1.1？我试图把由这个构建的jar文件放在kafka-连接-存储-共同目录中$CONFLUENT_HOME但我仍然得到同样的错误。

构建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 引用该类

如果有人成功地构建了自定义分区器，任何帮助都将不胜感激。

Answer 1

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

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

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

Answer 2

将自定义jar文件复制到插件文件夹Example/usr/local/share/Kafka/plugins/confluentinc-Kafka-connect-S3/下