正确配置 Kafka Connect S3 Sink TimeBasedPartitioner答案

【问题标题】：Properly Configuring Kafka Connect S3 Sink TimeBasedPartitioner正确配置 Kafka Connect S3 Sink TimeBasedPartitioner
【发布时间】：2018-06-16 03:39:30
【问题描述】：

我正在尝试使用 Confluent S3 接收器的 TimeBasedPartitioner。这是我的配置：

{  
"name":"s3-sink",
"config":{  
    "connector.class":"io.confluent.connect.s3.S3SinkConnector",
    "tasks.max":"1",
    "file":"test.sink.txt",
    "topics":"xxxxx",
    "s3.region":"yyyyyy",
    "s3.bucket.name":"zzzzzzz",
    "s3.part.size":"5242880",
    "flush.size":"1000",
    "storage.class":"io.confluent.connect.s3.storage.S3Storage",
    "format.class":"io.confluent.connect.s3.format.avro.AvroFormat",
    "schema.generator.class":"io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator",
    "partitioner.class":"io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "timestamp.extractor":"Record",
    "timestamp.field":"local_timestamp",
    "path.format":"YYYY-MM-dd-HH",
    "partition.duration.ms":"3600000",
    "schema.compatibility":"NONE"
}

}

数据是二进制的，我使用 avro 方案。我想使用实际的记录字段“local_timestamp”，它是一个 UNIX 时间戳来分区数据，比如每小时文件。

我使用通常的 REST API 调用启动连接器

curl -X POST -H "Content-Type: application/json" --data @s3-config.json http://localhost:8083/connectors

不幸的是，数据没有按照我的意愿进行分区。我还尝试删除冲洗尺寸，因为这可能会干扰。但是后来我得到了错误

{"error_code":400,"message":"Connector configuration is invalid and contains the following 1 error(s):\nMissing required configuration \"flush.size\" which has no default value.\nYou can also find the above list of errors at the endpoint `/{connectorType}/config/validate`"}%

知道如何正确设置 TimeBasedPartioner 吗？我找不到一个可行的例子。

另外，如何调试此类问题或进一步了解连接器实际在做什么？

非常感谢任何帮助或进一步的建议。

【问题讨论】：

标签： amazon-s3 apache-kafka apache-kafka-connect confluent-platform

【解决方案1】：

在研究TimeBasedPartitioner.java 的代码和日志之后

confluent log connect tail -f

我意识到时区和语言环境都是强制性的，尽管Confluent S3 Connector 文档中没有这样指定。以下配置字段解决了问题，让我将正确分区的记录上传到 S3 存储桶：

"flush.size": "10000",
"storage.class": "io.confluent.connect.s3.storage.S3Storage",
"format.class": "io.confluent.connect.s3.format.avro.AvroFormat",
"schema.generator.class": "io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator",
"partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
"path.format": "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
"locale": "US",
"timezone": "UTC",
"partition.duration.ms": "3600000",
"timestamp.extractor": "RecordField",
"timestamp.field": "local_timestamp",

还要注意两点：首先，flush.size 的值也是必需的，文件最终会被分区成更小的块，不大于flush.size 指定的值。其次，最好选择上面显示的 path.format，以便生成正确的树结构。

我仍然不能 100% 确定记录字段 local_timestamp 是否真的用于对记录进行分区。

非常欢迎任何 cmets 或改进。

【讨论】：

上述工作有效吗？我得到PartitionException: Error encoding partition 这可能意味着（根据您上面链接的代码）我的消息（Avro）没有正确反序列化。您是否使用了架构注册表？
是的，我使用了模式注册表并且上面的代码有效。我猜如果没有架构注册表，它就无法工作
谢谢丹尼尔。我还必须指定 "value.converter": "io.confluent.connect.avro.AvroConverter" 和 "value.converter.schema.registry.url": <my_schema_registry-url> 才能使其工作。默认情况下，转换器设置为ByteArrayConverter。所以，不得不重写它。
哦，是的，这是必需的，否则现在可以理解数据结构了。

【解决方案2】：

确实，您修改后的配置似乎是正确的。

具体来说，将timestamp.extractor 设置为RecordField 允许您根据您的记录所具有的时间戳字段以及您通过设置属性timestamp.field 来识别的时间戳字段对文件进行分区。

当设置timestamp.extractor=Record 时，基于时间的分区器将为每条记录使用 Kafka 时间戳。

关于flush.size，将此属性设置为较高的值（例如Integer.MAX_VALUE）实际上等同于忽略它。

最后，在最新版本的连接器中不再需要schema.generator.class。

【讨论】：