apache-kafka - 如何使用Kafka连接输出到GCS中的动态目录？

Question

我正在从 Kafka 主题中获取 JSON 数据。我需要将此数据转储到 GCS(Google Cloud Storage)到一个目录中，其中目录名称将从 JSON 数据中的“ID”值中获取。

我用谷歌搜索并没有找到任何类似的用例，其中 Kafka Connect 可用于解释 JSON 数据并根据来自 JSON 数据的值动态创建目录。
这可以使用 Kafka Connect 实现吗？

Answer 1

您可以使用 Kafka Connect GCS sink connector由 Confluent 提供。

The Google Cloud Storage (GCS) connector, currently available as a sink, allows you to export data from Kafka topics to GCS objects in various formats. In addition, for certain data layouts, GCS connector exports data by guaranteeing exactly-once delivery semantics to consumers of the GCS objects it produces.

这是连接器的示例配置:

name=gcs-sink
connector.class=io.confluent.connect.gcs.GcsSinkConnector
tasks.max=1
topics=gcs_topic

gcs.bucket.name=#bucket-name
gcs.part.size=5242880
flush.size=3

gcs.credentials.path=#/path/to/credentials/keys.json

storage.class=io.confluent.connect.gcs.storage.GcsStorage
format.class=io.confluent.connect.gcs.format.avro.AvroFormat
partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner

schema.compatibility=BACKWARD

confluent.topic.bootstrap.servers=localhost:9092
confluent.topic.replication.factor=1

# Uncomment and insert license for production use
# confluent.license=

您可以在我上面提供的链接中找到有关安装和配置的更多详细信息。