kafka topic数据如何实时同步写入mysql？

Question

已知kafka服务器：n9-kkt.test001.com:30902以及
topic：test.info.tst
如何实现topic数据实时同步写入mysql？

Answer 1

• 你可以参考下这篇文章：mysql通过kafka实现数据实时同步（一）——介绍
• 除此之外, 这篇博客: NiFi 学习 — kafka入库mysql中的 处理器 部分也许能够解决你的问题, 你可以仔细阅读以下内容或跳转源博客中阅读:
  

• 拉取一個EvaluateJsonPath 处理器，用来将kakfa内的json数据做简单的处理，主要是得到数据具体项目，具体配置如下：

  

  此操作的意义是 : 将json串中的数据条目，使用 $ 表达式的方式获取数据值,在数据走到这个处理器的时候，就可以拿到对应的数值。

• 以下回答由chatgpt基于相关博客总结生成:

  使用Kafka Connect将Kafka主题的数据实时同步写入MySQL的方法如下：

  1. 首先，下载kafka-connect-jdbc插件并解压。可以从https://www.confluent.io/hub/confluentinc/kafka-connect-jdbc 下载插件。

  2. 将插件中lib目录下的jar文件提取出来，并将其放入Kafka的libs目录。

  3. 复制Java的MySQL驱动包到Kafka的libs目录。

  4. 在kafka-connect-jdbc插件的etc目录中找到source.properties文件，并将其复制到Kafka的config目录下，并将文件名修改为connect-mysql-source.properties。

  5. 修改connect-mysql-source.properties文件，设置以下属性：

  6. name：连接器的名称，可以自定义。
  7. connector.class：连接器使用的类名。
  8. tasks.max：指定连接器运行的任务数量。
  9. connection.url：MySQL数据库的连接URL。
  10. table.whitelist：要同步的表名，以逗号分隔。

  除了上述属性，还可以根据需要设置其他属性，例如mode、timestamp.column.name等。

  1. 修改Kafka的config目录下的connect-standalone.properties文件，设置以下属性：
  2. bootstrap.servers：Kafka服务器的地址和端口。
  3. group.id：指定连接器所属的Connect集群的ID。
  4. key.converter：设置键转换器的类名。

  5. value.converter：设置值转换器的类名。

  6. 启动ZooKeeper和Kafka服务器。

  7. 启动Kafka Connect，执行以下命令： bin/connect-standalone.sh config/connect-standalone.properties config/connect-mysql-source.properties 注意：如果使用集群模式，需要修改connect-distributed.properties。

  8. 使用命令行工具消费Kafka主题，查看数据是否成功导入MySQL。 ./kafka-console-consumer.sh --bootstrap-server <kafka-server>:<kafka-port> --topic <mysql-topic> --from-beginning 其中，和是Kafka服务器的地址和端口，是导入MySQL的主题。

  这样，就可以实现将Kafka主题的数据实时同步写入MySQL。具体的配置和连接器属性可以根据实际情况进行调整和修改。