FLINKストリーミング：

Question

私のストリームは、キーがusr12345であり、値がストリーム出力に{"_key":"usr12345","_temperature":46.6}

.print()あるタイプTuple2<String,String>

.toString()出力(usr12345,{"_key":"usr12345","_temperature":46.6})

の記録を生産しているシリアル化された文字列のメッセージで予期しないcharaters正しく値：

(usr12345,{"_key":"usr12345","_temperature":46.6})

しかし、私はカフカにストリームを書き込む際にキーがusr12345（先頭に空白を含む）と値({"_key":"usr12345","_temperature":46.6}

お知らせスペースの先頭にキーと左括弧の初めになると値。

非常に奇妙です。なぜこれが起こるのでしょうか？

TypeInformation<String> resultType = TypeInformation.of(String.class); 

KeyedSerializationSchema<Tuple2<String, String>> schema = 
     new TypeInformationKeyValueSerializationSchema<>(resultType, resultType, env.getConfig()); 

FlinkKafkaProducer010.FlinkKafkaProducer010Configuration flinkKafkaProducerConfig = FlinkKafkaProducer010.writeToKafkaWithTimestamps(
     stream, 
     "topic",  
     schema, 
     kafkaProducerProperties); 

Answer 1

TypeInformationKeyValueSerializationSchema結果をバイナリデータとして解釈されなければならないことを意味するFLINKのカスタムシリアライザとデータをシリアル化：ここ

シリアル化コードです。 FlinkのStringシリアライザは、文字列の長さをすべての文字のエンコードに続けて書き込みます。

私はあなたがプレーンな文字列デシリアライザでカフカのトピックをデシリアライズすることを前提とします。キーの場合、シリアライズされた長さは空白文字として解釈されます。値の場合、長さは'('と解釈されます。

はプレーンな文字列としてキーと値をシリアライズまたは互換性のデシリアライザを使用する異なるシリアライザを使用してみてください。あなたが説明したことは少し奇妙です