我在我的flink stream工作中看到了一个奇怪的行为。<version>1.9.1</version>在我的pom中包含了这段代码后，我的代码停止了工作，并且我得到了异常：
com.esotericsoftware.kr
浏览 379提问于2020-05-24得票数 2
1回答
在我的Flink反序列化程序中，我从getProducedType方法中得到以下错误： at org.apache.flink.streaming.api.operators.AbstractStreamOperator:86)
at org.apache.flink.streami
浏览 31提问于2018-01-03得票数 3
回答已采纳
我正在尝试反序列化我的flink stream工作中的kafka事件。ConfluentRegistryAvroDeserializationSchema.forGeneric(schema, schemaRegistryURL), properties))Serialization(StreamSourceContexts.ja
浏览 249提问于2021-01-22得票数 1
我试图在ListState中初始化一个“CoProcessFunction”，但是它总是抛出这个错误“java.lang.NullPointerException:键控状态只能在‘键控流’上使用”。LOG.info("inside stream2 processor"); }抛出此错误的行是这一行stream1ListState = getRuntimeContext().getListState(
浏览 7提问于2021-03-11得票数 0
回答已采纳
我正在使用Flink来处理Kafka的流数据。流程是非常基本的，从Kafka消耗，数据丰富，然后沉到FS。偶尔，在一些Kafka分区中会出现延迟。重新启动应用程序可以帮助Flink“重新平衡”消费和延迟。然而，过了一段时间，我看到了其他分区的滞后等等。看到这种行为，我尝试使用Flink文档中建议的rebalance()来重新平衡消费比率：
“划分元素循环，为每个分区创建相同的负载。在存在数据倾斜的情况下，对于
浏览 4提问于2020-06-16得票数 2
回答已采纳
作为我的flink应用程序的运行时，我通过AWS ()使用托管flink "throwableInformation": [
"com.esotericsoftware.kryo.KryoException
浏览 2提问于2020-06-03得票数 1
回答已采纳
使用Apache 1.4.2我得到以下异常 at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:207)
at org.apache.flink.stream
浏览 0提问于2018-06-28得票数 1
回答已采纳
对于从保存点恢复Flink检查点，当我们使用java基本类型的值状态时，它可以正常运行，但当使用java对象的值状态时，它会抛出异常。它的关键解决方案是什么？任何想法都将受到感谢。at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext(StreamTask.java:738)
at org.apache.flink.streaming.runtime.tasks.Str
浏览 16提问于2019-05-29得票数 2
我的Flink作业运行了几天，没有任何问题，但是过了几天，就会关闭tm并重新启动整个作业。在日志中我发现，org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend -删除现有的实例基目录/tmp/flink-io-4b455efa-bcdeop_KeyedProcessOperator_a1c286a47e97622aa92a8f6cd4115854__1_4__uuid_4b53ff24-e240-48d6-b438-3ab2d05cbdb8
在
浏览 5提问于2020-10-06得票数 0
回答已采纳
我试图让Flink (1.2.0)在启用了HA的Hadoop集群(CDH 5.10.0)上工作，但当我通过终止活动RM来测试它时，它会导致整个集群宕机。我已经在flink-conf.yml中配置了Flink的HAhigh-availability.zookeeper.quorum: zookeeper1yarn.application-attempts: 2
然后，我使用yarn-session.sh -n 2 -nm "Flink HA test&qu
浏览 8提问于2017-03-08得票数 4
emptyDir: name: rocksdb-volume mountPath: /opt/flink/rocksdb at org.apache.flink.api.scala.typeutils.OptionSerializer.d
浏览 12提问于2022-02-08得票数 2
client, props); }它在我的机器上本地工作，但当我将它部署在yarn集群上时，我得到了以下异常： at org.apache.flink.streaming.runtime.tasks.SourceStreamTask: C
浏览 0提问于2020-01-23得票数 1
deserialize message from [akka://application@127.0.0.1:2551] with serializer id [123454323] and manifest []. com.esotericsoftware.kryo.KryoException: java.lang.NullPointerException
浏览 402提问于2020-12-01得票数 0
我正在尝试做一个POC的Flink状态模式演变。我使用的是Flink 1.15.0和Java 11。(BackendRestorerProcedure.java:135)Caused by: com.esotericsoftware.kryo.KryoExceptioncom.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:113)我预计io.peleg.kryo.User会抛出
浏览 66提问于2022-07-02得票数 2
2回答
我试图用Flink编写一个计算，它需要两个阶段。在第二阶段，我想使用这个对象来生成模拟数据。这是我得到的例外： at org.apache.flink.api.java.io.LocalCollectionOutputFormat.closeThe program execution failed: <
浏览 3提问于2015-04-30得票数 4
回答已采纳
3回答
at org.apache.flink.streaming.api.operators.StreamTaskStateInitializerImpl.streamOperatorStateContext:415) at org.apache.flink.runtime.taskmanager.Task.run(BackendRestorerProcedure.java:121)
浏览 31提问于2020-11-05得票数 0
回答已采纳
</groupId> <version>1.1.1</version> <groupId>org.apache.flink</groupId>
<artifactId>flink
浏览 2提问于2016-09-03得票数 3
回答已采纳
1回答
我在flink中使用广播状态模式，在这里我试图连接两个流，一个流是规则的控制流，另一个流是整数流(用于虚拟播放目的)。我尝试了以下步骤来检查模式演化是否适用于此，如docs中提到的那样：提交作业jar停止工作。尝试使用flink -s <savepoint>命令还原作业。这样，作业就无法重新启动，因为模式演化失败，导致以下错误：
由: org.apache.flink.util.Fl
浏览 3提问于2022-05-06得票数 0
context.getKeyedStateStore().getMapState(descriptor);当我将“描述符”分配给getMapState()时，我得到了getMapState()java.lang.NullPointerException(AbstractUdfStreamOperator.java:96)
at org.apache.flink.streaming.api.operators.AbstractStreamOperator.initializeState(AbstractStreamOperator.j
浏览 0提问于2018-03-18得票数 0
import com.twitter.chill.protobuf.ProtobufSerializer;import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.scala.hadoop.mapreduce.HadoopInputFormat> dataSet = ee.createInput(hadoopIF).s
浏览 6提问于2019-01-21得票数 0
云服务器
ICP备案
云直播
即时通信 IM
对象存储
腾讯云开发者
