用canal同步binlog到kafka，sparkstreaming消费kafkatopic乱码问题

canal 1.1.1版本之后, 默认支持将canal server接收到的binlog数据直接投递到MQ, 目前默认支持的MQ系统有kafka和RocketMQ。

在投递的时候我们使用的是非压平的消息模式（canal.mq.flatMessage =false //是否为flat json格式对象），然后消费topic的时候就一直无法正常显示和序列化，通过kafka-console-consumer.sh命令收到的消息如下图

在github上也能找到相关问题

canal-kafka 数据同步到kafka之后，kafka topic乱码：https://github.com/alibaba/canal/issues/898

canal.kafka 用bin/kafka-console-consumer.sh命令收到乱码：https://github.com/alibaba/canal/issues/1013

 在非flatmessage模式下向kafka数据投递传输的是数据包，收到数据后还要解包成对应的message，可参考canal client中的kafka实现, github地址为 https://github.com/alibaba/canal/tree/master/client/src/main/java/com/alibaba/otter/canal/client/kafka

打开连接后 kafkaCOnsumer= new KafkaConsumer(properties);

参考这种操作只是简单的kafka能够收消息，结合spark streaming收消息也差不多。

在kafkaparam中设置key和value的反序列化方式

"key.deserializer" -> classOf[StringDeserializer].getName
"value.deserializer" -> classOf[MessageDeserializer].getName

 在拉取消息的时候设置接受格式为Array[Byte]

val messages = KafkaUtils.createDirectStream[String, Array[Byte], StringDecoder, DefaultDecoder](ssc, kafkaParams, topics)

在处理每个RDD的时候再对内容进行反序列化：

val parData = rdd.mapPartitions(t => {
val mesDesc = new MessageDeserializer
var list = List[consumerUser]()
while (t.hasNext) {
try {
val value = t.next()._2
val message = mesDesc.deserialize("", value)
//val listMaps = CanalParse.parseData(message)
//逻辑
} catch {
case e: Exception => log.error(e)
}
}
list.iterator
})

这样就拿到了message对象。

依赖jar包

    org.apache.spark
    spark-streaming-kafka_2.11
    1.6.3

    org.apache.kafka
    kafka_2.11
    0.8.2.1


    com.alibaba.otter
    canal.protocol
    1.1.0


    com.alibaba.otter
    canal.kafka.client
    1.1.0