hadoop - Confluent HDFS 连接器

Question

我想将 kafka 日志文件移动到 hadoop 日志文件。所以我遵循 HDFS 连接器配置

/quickstart-hdfs.properties

name=hdfs-sink 
connector.class=io.confluent.connect.hdfs.HdfsSinkConnector
tasks.max=1 
topics=kafka_log_test 
hdfs.url=hdfs://10.100.216.60:9000 
flush.size=100000 
hive.integration=true 
hive.metastore.uris=thrift://localhost:9083 
schema.compatibility=BACKWARD 
format.class=io.confluent.connect.hdfs.parquet.ParquetFormat 
partitioner.class=io.confluent.connect.hdfs.partitioner.Hour‌​lyPartitioner

/connect-avro-standalone.properties

bootstrap.servers=localhost:9092

key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://localhost:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://localhost:8081

internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

offset.storage.file.filename=/tmp/connect.offsets

当我运行 HDFS 连接器时，只需在 .avro 文件中写入 avro 模式。不是数据。

/kafka_log_test+0+0000000018+0000000020.avro

 avro.schema {"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}],"connect.version":1,"connect.name":"myrecord"}

主题有大量数据，但 confluent hdfs 连接器不会将数据移动到 hdfs。

我该怎么做才能解决这个问题？

Answer 1

根据定义，除非消息在偏移量 18 和 20 之间以其他方式压缩或过期，否则包含名称 0+0000000018+0000000020 的文件将从分区 0 中获得 2 条记录。

您应该使用avro-tools 的tojson 命令，而不是getmeta。

或者您可以使用 Spark 或 Pig 来读取该文件。

---

您可能还想验证连接器在启动后是否继续运行，因为在不是 Hive Metastore 服务器的机器上设置 hive.metastore.uris=thrift://localhost:9083将导致连接任务失败。 URI 应该是 Hive 的实际主机，就像您为 NameNode 所做的那样。

---

另外，不管怎么说，用 format.class=io.confluent.connect.hdfs.parquet.ParquetFormat 应该不可能得到一个 .avro 文件扩展名，所以你可能想验证你正在寻找正确的 HDFS 路径。注意:在写入最终输出文件之前，Connect 会暂时写入 +tmp 位置。