使用kafka connect的hdfs中没有avro数据

Question

我正在使用Kafka连接分布。命令是：bin/connect-distributed etc/schema-registry/connect-avro-distributed.properties

工作人员配置为：

    bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
    group.id=connect-cluster
    key.converter=org.apache.kafka.connect.json.JsonConverter
    value.converter=org.apache.kafka.connect.json.JsonConverter
    key.converter.schemas.enable=false
    value.converter.schemas.enable=false

Kafka连接重新开始没有错误！

    curl -X POST -H "Content-Type: application/json" --data '{"name":"hdfs-sink-mysiteview","config":{"connector.class":"io.confluent.connect.hdfs.HdfsSinkConnector","tasks.max":"3","topics":"mysiteview","hdfs.url":"hdfs://master1:8020","topics.dir":"/kafka/topics","logs.dir":"/kafka/logs","format.class":"io.confluent.connect.hdfs.avro.AvroFormat","flush.size":"1000","rotate.interval.ms":"1000","partitioner.class":"io.confluent.connect.hdfs.partitioner.DailyPartitioner","path.format":"YYYY-MM-dd","schema.compatibility":"BACKWARD","locale":"zh_CN","timezone":"Asia/Shanghai"}}'  http://kafka1:8083/connectors

    {"f1":"192.168.1.1","f2":"aa.example.com"}

java代码如下:

Properties props = new Properties();
props.put("bootstrap.servers","kafka1:9092");
props.put("acks","all");
props.put("retries",3);
props.put("batch.size", 16384);
props.put("linger.ms",30);
props.put("buffer.memory",33554432);
props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<String,String>(props);
Random rnd = new Random();
for(long nEvents = 0; nEvents < events; nEvents++) {
    long runtime = new Date().getTime();
    String site = "www.example.com";
    String ipString = "192.168.2." + rnd.nextInt(255);
    String key = "" + rnd.nextInt(255);
    User u = new User();
    u.setF1(ipString);
    u.setF2(site+" "+rnd.nextInt(255));
    System.out.println(JSON.toJSONString(u));
    producer.send(new ProducerRecord<String,String>("mysiteview",JSON.toJSONString(u)));
    Thread.sleep(50);
}

producer.flush();
producer.close();

奇怪的事情发生了。我从kafka-logs中获取数据，但在hdfs中没有数据（没有主题目录）。我尝试connector命令：

    curl -X GET http://kafka1:8083/connectors/hdfs-sink-mysiteview/status

    {"name":"hdfs-sink-mysiteview","connector":{"state":"RUNNING","worker_id":"10.255.223.178:8083"},"tasks":[{"state":"RUNNING","id":0,"worker_id":"10.255.223.178:8083"},{"state":"RUNNING","id":1,"worker_id":"10.255.223.178:8083"},{"state":"RUNNING","id":2,"worker_id":"10.255.223.178:8083"}]}

    curl -X GET http://kafka1:8083/connectors/hdfs-sink-mysiteview/hdfs-sink-siteview-1

出什么问题了？

Answer 1

在没有看到Worker日志的情况下，我不确定在使用上面描述的设置时HDFS连接器实例到底有哪一个异常失败。不过，我可以发现配置中的一些问题：

1. 您提到使用bin/connect-distributed etc/schema-registry/connect-avro-distributed.properties启动连接工作者。这些属性默认将键和值转换器设置为avroconverter，并要求您运行schema-registry服务。如果您确实已经编辑了connect-avro-distributed.properties中的配置，以使用JSONConverter作为替代，那么在将Kafka记录转换为connect的sinkrecord数据类型期间，HDFS连接器可能会失败，就在它试图将数据导出到HDFS之前。
2. 直到最近，HDFS连接器只能将Avro记录导出到Avro或Parquet格式的文件中。这需要使用前面提到的avroconverter。最近添加了将记录作为JSON导出到文本文件的功能，该功能将出现在连接器的4.0.0版本中（您可以通过签出并从源代码处构建连接器来尝试该功能）。

此时，我的第一个建议是尝试使用bin/kafka-avro-console-producer导入数据。定义它们的模式，确认使用bin/kafka-avro-console-consumer成功导入了数据，然后将HDFS连接器设置为使用avroformat。连接器页面上的quickstart描述了一个非常相似的过程，也许这将是您用例的一个很好的起点。