我正在使用Flink，我正在使用Kafka连接器。我从flink收到的消息是一个逗号分隔的项目列表。"'在我的例子中，我必须编写自己的反序列化程序来实现DeserializationSchema并返回一个元组或Pojo。所以基本上用我自己的函数替换SimpleStringSchema()。Flink提供了一些开箱即用的反序列化
浏览 69提问于2020-10-24得票数 1
回答已采纳
我正在使用flink 1.4.2从Kafka读取数据，并使用JSONDeserializationSchema将它们解析为JSONDeserializationSchema。如果传入的记录不是有效的JSON，那么我的Flink作业就会失败。我想跳过破纪录而不想失败。DataStream<ObjectNode> messageStream = env.addSource(kafkaConsumer);如果Kafka中的数据不是有效
浏览 1提问于2018-07-12得票数 3
回答已采纳
我在使用flink的SerializationSchema时遇到了一些困难。Sink函数：import org.apache.flink.streaming.api.functions.sink.SinkFunctionSource函数(不确定它对我遇到的问题是否有用)：
import org.apache.<e
浏览 4提问于2020-09-02得票数 0
回答已采纳
2回答
自定义类class Person private Integer id; //getters and setters卡夫卡Flink连接器TypeInformationSerializationSchemapersons", schema , getKafkaProperties()));
现在，如果我发送下
浏览 1提问于2018-07-02得票数 0
回答已采纳
1回答
我们试图在KafkaAvroSerializer中使用Server连接器，并在将数据发送给Kafka之前提供自定义的ProducerInterceptor来加密数据。在使用者端，我们希望使用Flink连接器解密，然后使用适当的反序列化器。为了实现这一点，我们提出了几个问题：
1)如果我们提供定制的ConsumerInterceptor来解密数据，那么在Flink中创建DataStream时，应该通过属性文件传递这些数据吗？, properties));
浏览 0提问于2018-04-20得票数 0
回答已采纳
我有一个带有以下Avro IDL的Kafka主题，并注册到Schema Registry。(AvroRowDataDeserializationSchema.java:46) org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:82) org.apache.flink<
浏览 190提问于2021-10-12得票数 1
回答已采纳
首先，我在Kafka topic中加载了一个CSV文件，我可以通过中的Flink程序打印该主题。(); stream.print();
env.execute("FlinkStreaming Java API Skeleton"); 我的问题是，我想逐行阅读主题，并分别处理每一行，您能指导我如何逐行阅读卡夫卡主题吗？
浏览 16提问于2019-06-24得票数 0
回答已采纳
at org.apache.flink.runtime.io.network.channels.InputChannel.readRecord(InputChannel.java:158) at org.apache.flink.runtime.io.network.api.MutableRecordRe
浏览 0提问于2015-05-13得票数 1
1回答
我正在Scala中构建一个Apache应用程序，它从Kafka总线读取流数据，然后对其执行汇总操作。Kafka的数据是Avro格式的，需要一个特殊的反序列化类。String] <and>
(x$1: java.util.regex.Pattern,x$2: org.apache.flink.api.common.serialization.DeserializationSchemaString] <and>
浏览 1提问于2018-07-01得票数 0
回答已采纳
我是新的Flink和做一些非常类似于下面的链接。
我还试图添加JSONDeserializationSchema()作为我的Kafka输入JSON消息的反序列化器，该消息没有键。
浏览 4提问于2020-05-28得票数 4
回答已采纳
场景: Apache Flink、Kafka、Protocol buffers data consumer。数据来源为协议缓冲区格式的Kafka主题(多个主题: topic#1、topic#3、topic#3)。使用者是Apache Flink使用者。每个主题都有一个独特的协议定义。Flink中开发一个通用的数据摄取作业，将Kafka中的数据摄取到数据库中。如何为Apache Flink</
浏览 13提问于2021-04-03得票数 0
3回答
关于Flink反序列化的两个问题
、、、、
我只想在Flink中检索它们，然后用不同的键分离消息。我试过所有的去序列化器，但都没有用。简单的反序列化器不明显地起作用，但它完全忽略键。(4/4)在org.apache.flink.runtime.util.DataInputDeserializer.readUnsignedByte(DataInputDeserializer.java:306)切换到失败
浏览 8提问于2017-06-12得票数 1
3回答
我使用flink从kafka读取数据并将其转换为protobuf。我面临的问题是，当我运行java应用程序时，会出现以下错误。我还试图在阅读kafka时直接反序列化，但我不确定getProducedType()方法应该返回什么。(TypeExtractor.java:1515)已经包含在类的层次结构中，在org.apache.flink.api.java.typeutils.TypeExtractor.analyzePojo(TypeExtractor.java:151
浏览 2提问于2016-07-09得票数 2
我正在创建一个简单的Apache作业(使用Scala)，它只尝试打印表示RabbitMQ队列(RMQSource)接收的事件的case类。我已经创建了自己的反序列化模式(使用Jackson)，当使用的消息实际上是一个表示case类的JSON时，它可以正常工作。但是，如果队列接收到格式错误的事件(我猜，我们可以称它为“中毒消息”)，作业就会失败并继续重新启动。我必须清除队列，然后作业状态更改为“运行”。当收到有毒信息时，<em
浏览 1提问于2021-11-16得票数 1
回答已采纳
我以为所有的Java依赖都是和pip install apache-flink一起安装的 以上错误发生在该行上： kafka_consumer = FlinkKafkaConsumer(
topics
浏览 252提问于2021-09-26得票数 0
我无法让它使用不推荐的构造函数序列化数据(这并不奇怪)，我也无法在网上找到任何关于如何实现序列化器的实现或示例(所有示例都使用旧的Kafka连接器)FlinkKafkaProducer010<init>(FlinkKafkaProducer.java:525)
at org.apache.flink.stream
浏览 1提问于2019-10-31得票数 6
回答已采纳
相关问题：https://issues.apache.org/jira/browse/FLINK-14380 在使用类型提取器Flink判断一个类是否符合POJO时，Flink会检查该类是否实现了setter这个问题阻碍了不可变化性，并使得在不使用Kryo序列化程序的情况下无法使用case类。问题位于https://github.com/apache&#x
浏览 37提问于2019-10-12得票数 2
at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleaner.java:99) at org.apache.flink.streaming.api.datastream.Da
浏览 0提问于2018-04-18得票数 0
2回答
我正在尝试使用AVRO 1.0.3从Kafka主题读取Flink数据。import org.apache.flink.api.java.typeutils.TypeExtractor; reader = n
浏览 3提问于2016-08-02得票数 3
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
云直播活动推荐
腾讯云开发者
