【发布时间】:2016-07-02 09:09:54
【问题描述】:
我正在尝试使用 Confluent 平台提供的 kafka-hdfs-connector 将数据从 Kafka 复制到 Hive 表中。虽然我能够成功地做到这一点,但我想知道如何根据时间间隔存储传入的数据。例如,我希望每 5 分钟创建一个新分区。
我用 partition.duration.ms 尝试了 io.confluent.connect.hdfs.partitioner.TimeBasedPartitioner 但我认为我做错了。我在 Hive 表中只看到一个分区,所有数据都进入该特定分区。像这样的:
hive> show partitions test;
OK
partition
year=2016/month=03/day=15/hour=19/minute=03
所有的 avro 对象都被复制到这个分区中。
相反,我想要这样的东西:
hive> show partitions test;
OK
partition
year=2016/month=03/day=15/hour=19/minute=03
year=2016/month=03/day=15/hour=19/minute=08
year=2016/month=03/day=15/hour=19/minute=13
最初连接器将创建路径 year=2016/month=03/day=15/hour=19/minute=03 并将继续将所有传入数据复制到此目录中以供接下来 5分钟,并在第 6 分钟开始时创建一个新路径,即 year=2016/month=03/day=15/hour=19/minute=08 并复制下一个 5 的数据分钟进入此目录,依此类推。
这就是我的配置文件的样子:
name=hdfs-sink
connector.class=io.confluent.connect.hdfs.HdfsSinkConnector
tasks.max=1
topics=test
hdfs.url=hdfs://localhost:9000
flush.size=3
partitioner.class=io.confluent.connect.hdfs.partitioner.TimeBasedPartitioner
partition.duration.ms=300000
path.format='year'=YYYY/'month'=MM/'day'=dd/'hour'=HH/'minute'=MM/
locale=en
timezone=GMT
logs.dir=/kafka-connect/logs
topics.dir=/kafka-connect/topics
hive.integration=true
hive.metastore.uris=thrift://localhost:9083
schema.compatibility=BACKWARD
如果有人能指出我正确的方向,那将非常有帮助。如果需要,我很乐意分享更多细节。不想让这个问题看起来像一个永无止境的问题。
非常感谢!
【问题讨论】:
标签: hadoop hive apache-kafka kafka-consumer-api kafka-producer-api