Kafka для передачи данных в базу данных
I имеет данные, поступающие в Kafka, я хочу передать данные из kafka в базу данных (Postgresql).
Я следую шагам по этой ссылке "https://hellokoding.com/kafka-connect-sinks-data-to-postgres-example-with-avro-schema-registry-and-python/" и получаю сообщение об ошибке. Какие-либо предложения.
Ответы 2
Это не требует кода, этого можно добиться с помощью Kafka connect и коннектора mysql, доступных в Confluent.
https://www.confluent.io/hub/confluentinc/kafka-connect-jdbc
Apache Kafka имеет лицензию Apache 2.0 и включает Kafka Connect. Коннектор JDBC SInk имеет лицензию сообщества (confluent.io/confluent-community-license-faq) и, как правило, может использоваться бесплатно.
Конечно, большое спасибо. Еще одна вещь. Существуют разные источники данных, из которых данные поступают в Kafka через mqtt. Я хочу извлечь эти данные из темы Kafka в базу данных, так как мне это получить. @РобинМоффат
@RobinMoffatt Я настроил Postgresql на локальном хосте, и я пытаюсь отправить сообщение в тему, но таблица все еще не создана, следую этой и более поздней статье, docs.confluent.io/current/connect/kafka-connect-jdbc/… и hellokoding.com/…
Почему нет данных?: confluent.io/blog/…
Я обновил свой вопрос скриншотом ошибки, которую я получаю, какая-либо помощь из этого ?? @РобинМоффат
Конечно, я позабочусь об этом в следующий раз, а пока есть мысли о том, что может быть не так @RobinMoffatt
Да, я разместил ответ.
Ошибка, которую вы получаете, заключается в следующем:
Failed to deserialize the data for topic
Error deserializing Avro message
Unknown magic byte
Это потому, что вы указали конвертер Avro, но данные в вашей теме не Avro.
Подробнее см. в этой статье: https://www.confluent.io/blog/kafka-connect-deep-dive-converters-serialization-explained.
Я обновил свой код, и я могу получить данные. Но поток, который я хочу, - это прямо из темы Kafka в базу данных, я хочу пропустить этот код Python, возможно ли это?
Я получаю данные в kafka через mqtt по этой ссылке «github.com/kaiwaehner/kafka-connect-iot-mqtt-connector-example/…», поэтому мой формат темы kafka похож на mqtt.topicname, но этот формат не принимается текущим кодом python. Я получаю сообщение об ошибке, когда я использую эту тему.
для нового вопроса/проблемы начните новый вопрос.
да, но это не бесплатно. Я взял трейловый период. Неужели нет другого способа сделать это.