canal使用非flatmessage方式獲取mysql bin log日志發(fā)至kafka比直接發(fā)送json效率要高很多，數(shù)據(jù)發(fā)到kafka后需要實(shí)時(shí)解析為json,這里可以使用strom或者flink,公司本來(lái)就是使用strom解析，但是在吞吐量上有瓶頸，優(yōu)化空間不大。所以試一試通過(guò)flink來(lái)做。

非flatmessage需要使用特定的反序列化方式來(lái)處理為Message對(duì)象，所以這里需要自定義一個(gè)類

1 /** 2 * 反序列化canal binlog 3 * 4 * @author @ 2019-02-20 5 * @version 1.0.0 6 */ 7 @PublicEvolving 8 public class MessageDeserializationSchema implements KeyedDeserializationSchema<Message> { 9 10 private static final long serialVersionUID = -678988040385271953L; 11 private MessageDeserializer mesDesc; 12 13 @Override 14 public Message deserialize(byte[] messageKey, byte[] message, String topic, int partition, long offset) throws IOException { 15 try { 16 if (mesDesc == null) { 17 mesDesc = new MessageDeserializer(); 18 } 19 Message result = mesDesc.deserialize(topic, message); 20 //result.setMetaData(topic, partition, offset); 21 return result; 22 } catch (Exception e) { 23 System.out.println(e); 24 } 25 return null; 26 } 27 28 @Override 29 public boolean isEndOfStream(Message nextElement) { 30 return false; 31 } 32 33 @Override 34 public TypeInformation<Message> getProducedType() { 35 return getForClass(Message.class); 36 } 37 }

?

然后就可以獲取到DataStream[Message]，但是在做算子操作的時(shí)候就報(bào)錯(cuò)了,意思是不支持kryo序列化

com.esotericsoftware.kryo.KryoException: java.lang.UnsupportedOperationException Serialization trace: props_ (com.alibaba.otter.canal.protocol.CanalEntry$Header) header_ (com.alibaba.otter.canal.protocol.CanalEntry$Entry) entries (com.alibaba.otter.canal.protocol.Message)at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)at com.esotericsoftware.kryo.Kryo.readObjectOrNull(Kryo.java:730)at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:113)at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)at com.esotericsoftware.kryo.Kryo.readObjectOrNull(Kryo.java:730)at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:109)at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:22)at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)at com.esotericsoftware.kryo.serializers.FieldSerializer.read(FieldSerializer.java:528)at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:657)at org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer.copy(KryoSerializer.java:231)at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:577)at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:554)at org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:534)at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:718)at org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:696)at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104)at org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collectWithTimestamp(StreamSourceContexts.java:111)at org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordWithTimestamp(AbstractFetcher.java:398)at org.apache.flink.streaming.connectors.kafka.internal.Kafka010Fetcher.emitRecord(Kafka010Fetcher.java:89)at org.apache.flink.streaming.connectors.kafka.internal.Kafka09Fetcher.runFetchLoop(Kafka09Fetcher.java:154)at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:665)at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:94)at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:58)at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.run(SourceStreamTask.java:99)at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)at org.apache.flink.runtime.taskmanager.Task.run(Task.java:704)at java.lang.Thread.run(Thread.java:748) Caused by: java.lang.UnsupportedOperationExceptionat java.util.Collections$UnmodifiableCollection.add(Collections.java:1055)at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:109)at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:22)at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:679)at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:106)... 29 more

?

參考官方文檔，需要注冊(cè)類的序列化方式：https://ci.apache.org/projects/flink/flink-docs-release-1.7/dev/custom_serializers.html

　　//message 不支持kryo序列化 不然在map flatmap的時(shí)候報(bào)錯(cuò)

　　env.getConfig.addDefaultKryoSerializer(classOf[Message], classOf[StringSerializer])

如果在算子之間會(huì)有其他對(duì)象傳輸?shù)脑?#xff0c;也同樣需要注冊(cè)。最后通過(guò)測(cè)試，flink解析的量大概在單個(gè)solt 1W+/s 左右。

?

轉(zhuǎn)載于:https://www.cnblogs.com/createweb/p/10580281.html

總結(jié)

以上是生活随笔為你收集整理的Flink解析kafka canal未压平数据为message报错的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。