java - Kafka 流消费为 CSV

Question

我在java中使用Kafka，我将JSON消息作为字符串使用，速率为每分钟100万条消息，我需要将字符串拆分为仅取一些值并将其保存到CSV以将其加载到数据库，我怎样才能制作这样的东西？

Answer 1

您可以使用 Kafka Connect JDBC sink 将数据从 Kafka 主题直接流式传输到数据库。要了解有关 Kafka Connect 的更多信息，请参阅 the docs和 this talk 。

以下是接收器连接器配置示例:

{
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "connection.url": "jdbc:mysql://mysql:3306/demo",
    "topics": "test01",
    "connection.user": "connect_user",
    "connection.password": "asgard",
    "auto.create": true,
    "auto.evolve": true,
    "insert.mode": "insert",
    "pk.mode": "record_key",
    "pk.fields": "MESSAGE_KEY"
}

在本教程中了解更多信息:https://rmoff.dev/kafka-jdbc-video

了解如何在 Kafka Connect 中安装 JDBC 驱动程序 here