GP
т.е. все работает так?
Size: a a a
2021 January 22
A
StreamRecord<T> copy = castRecord.copy(serializer.copy(castRecord.getValue()));
RK
блин, ну я пытался найти примере в интернете
A
вот это отваливается
RK
и делал как в примерах
RK
StreamRecord<T> copy = castRecord.copy(serializer.copy(castRecord.getValue()));
спасибо), и что делать?
RK
куды бечь?)
GP
блин, ну я пытался найти примере в интернете
ну если работает когда у тебя авросехеме не валом, то проблема в сериализации
GP
я немного удивлен что def schema: Schema = AvroSchema[URLResponse] не работает
GP
а вообще зачем нужен AvroSchema?
там есть AvroSerializationSchema который у флинка есть
там есть AvroSerializationSchema который у флинка есть
GP
т.е. либа не нужна
RK
я немного удивлен что def schema: Schema = AvroSchema[URLResponse] не работает
пробовал, тот же эффект
GP
может у нее лучше с сериализацией
A
@pomadchin
https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java#L259-L275
там веселье что сериализованное сообщение потом не может десериализовать
https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java#L259-L275
там веселье что сериализованное сообщение потом не может десериализовать
A
executionConfig.disableForceKryo()
вы уверены что вам это нужно?
вы уверены что вам это нужно?
GP
@pomadchin
https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java#L259-L275
там веселье что сериализованное сообщение потом не может десериализовать
https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/java/typeutils/runtime/kryo/KryoSerializer.java#L259-L275
там веселье что сериализованное сообщение потом не может десериализовать
ну понятное дело
GP
ну я так понял что вот это работает
val stream = env.addSource(new FlinkKafkaConsumer[GenericRecord](
kafkaTopic, ConfluentRegistryAvroDeserializationSchema.forGeneric(AvroSchema[URLResponse], schemaRegistryURL), properties))
GP
но может я криво прочитал)
RK
https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kafka.html#the-deserializationschemaRK
тут описаны несколько способов