Блог

Есть примеры и документация по копированию данных из разделов Kafka в S3, но как вы копируете данные из S3 в Kafka?

Когда вы читаете объект S3, вы получаете поток байтов. И вы можете отправить любой массив байтов в Kafka с ByteArraySerializer .

Или вы можете проанализировать этот входной поток в какой-либо пользовательский объект, а затем отправить его с помощью любого сериализатора, который вы можете настроить.

Здесь вы можете найти один пример процесса Kafka Connect (который, я полагаю, вы сравниваете с программой Confluent S3 Connect writer) —https://jobs.zalando.com/tech/blog/backing-up-kafka-zookeeper/index.html это может быть настроено для чтения двоичных архивов или текста, разделенного строками, из S3.

Аналогично, Apache Spark, Flink, Beam, NiFi и т.д. Инструменты, связанные с simlar Hadoop, также могут считывать данные из S3 и записывать события в Kafka.

Проблемы с этим подходом заключаются в том, что вам нужно отслеживать, какие файлы были прочитаны до сих пор, а также обрабатывать частично прочитанные файлы.

В зависимости от вашего сценария или желаемой частоты загрузки объектов, вы можете использовать лямбда-функцию для каждого события (например, при каждой загрузке файла) или как cron. Этот лямбда-код работает как производитель с использованием API Kafka и публикуется в теме.

Особенности:

• Триггером для функции Lambda может быть s3:PutObject событие, исходящее непосредственно из s3 или из событий cloudwatch.
• Вы можете запустить lambda как cron, если вам не нужны объекты мгновенно. Альтернативой в этом случае также может быть запуск cron на экземпляре EC2, у которого есть Kafka producer и разрешения на чтение объектов из s3, и он продолжает отправлять их в разделы kafka.