Как использовать Kafka connect для вывода в динамический каталог в GCS?

Question

Я получаю данные JSON из темы Кафки.Мне нужно выгрузить эти данные в GCS (Google Cloud Storage) в каталог, где имя каталога будет выбрано из значения «ID» в данных JSON.

Я гуглил и не нашел ни одного подобного варианта использования, в котором Kafka Connect можно использовать для интерпретации данных JSON и динамического создания каталогов на основе значения из данных JSON.Можно ли этого добиться с помощью Kafka Connect?

Answer 1

Вы можете использовать Разъем приемника Kafka Connect GCS , предоставляемый Confluent.

Соединитель Google Cloud Storage (GCS), доступный в настоящее время как приемник, позволяет экспортировать данные из тем Кафки в объекты GCS в различные форматы. Кроме того, для определенных макетов данных, разъем GCS экспортирует данные, гарантируя семантику доставки точно потребители объектов GCS, которые он производит.

Вот пример конфигурации для разъема:

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=

Подробнее об установке и настройке вы можете найти по ссылке, указанной выше.

Answer 2

На самом деле это невозможно из коробки при использовании большинства разъемов. Вместо этого вы можете реализовать свою собственную задачу приемника Kafka Connect, которая обрабатывает записи Kafka и затем записывает их в правильные каталоги GCS на основе вашего JSON.

Вот метод, который вы бы переопределили в разъеме.

Вот ссылка на исходный код для разъема приемника AWS S3.