Kafka乱码——中文单词在debezium中变成乱码

Posted 2023-03-23

【中文标题】Kafka乱码——中文单词在debezium中变成乱码

【英文标题】：Kafka messy code - Chinese words become messy code in debezium

【发布时间】：2019-03-21 02:57:45

【问题描述】：

我的 debezium 代码很乱：

"doulist_name": "2013 豆瓣电影��碑榜】" 

mysql数据库有中文单词，我用debezium把数据发给kafka。消费消息时发现中文单词乱码，请问如何解决？有什么可以使用的配置吗？

当我使用flume和kafka producer生成中文单词时，效果很好

部分配置：

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true
internal.key.converter=org.apache.kafka.connect.json.JsonConverter
internal.value.converter=org.apache.kafka.connect.json.JsonConverter
internal.key.converter.schemas.enable=false
internal.value.converter.schemas.enable=false

connector.class=io.debezium.connector.mysql.MySqlConnector
database.server.id=18405
database.server.name=mysqlfullfillment
database.whitelist=test
database.history.kafka.bootstrap.servers=192.168.0.100:9092
database.history.kafka.topic=dbhistory.fullfillment-local
include.schema.changes=true
transforms=unwrap
transforms.unwrap.type=io.debezium.transforms.UnwrapFromEnvelope

mysql 字符集：utf8 mysql config picture

版本：debezium v​​0.7.5、kafka v1.1.1

添加：

当我用 console./kafka-console-consumer.sh --zookeeper 192.168.0.100:2181 --topic mysqlfullfillment.test.doulist 测试它时 我的代码很乱

"doulist_name": "2013 豆瓣电影��碑榜】"

在我的 spark 代码中，我得到了同样混乱的代码：

  def main(args: Array[String]) 
    val spark = SparkSession
      .builder()
      .master("local")
      .appName("KafkaWordCount")
      .config("spark.streaming.stopGracefullyOnShutdown", "true")
      .getOrCreate()
    simpleTestCode(spark)
  

  def simpleTestCode(spark: SparkSession): Unit = 
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "localhost:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "KafkaWordCountgroup",
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (true: java.lang.Boolean)
    )
    val topics = Array("mysqlfullfillment.test.doulist")
    val ssc = new StreamingContext(spark.sparkContext, Seconds(2))

    ssc.checkpoint("/home/feng/software/code/bigdata/spark-warehouse")
    val stream = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferBrokers,
      Subscribe[String, String](topics, kafkaParams)
    )

    stream.map(mapFunc = record => (record.key, record.value)).foreachRDD(
      r => r.collect().foreach(t => print("message:" + t)))

    ssc.start()
    ssc.awaitTermination()
  

【问题讨论】：

这可能是字符编码问题。您究竟是如何使用来自 kafka 的消息的？您可以在上面的问题中添加该代码吗？

控制台消费者只执行UTF8解码

【参考方案1】：

我解决了这个问题。

当我在 debezium 中使用 JsonConverter 时：

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter

它会使用JsonSerializer来序列化数据，所以我必须在kafka中使用JsonDeserializer

val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> CommonUtil.getKafkaServers,
      "key.deserializer" -> classOf[JsonDeserializer],
      "value.deserializer" -> classOf[JsonDeserializer],
      "group.id" -> groupId,
      "auto.offset.reset" -> "latest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )