gpt4 book ai didi

mysql - Kafka乱码——Debezium中中文单词变成乱码

转载 作者:行者123 更新时间:2023-11-29 16:50:26 24 4
gpt4 key购买 nike

我用 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

添加:

当我用控制台测试它时./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()
}

最佳答案

我解决了这个问题。

当我在 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)
)

关于mysql - Kafka乱码——Debezium中中文单词变成乱码,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/52837776/

24 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com