Flink streaming:如何控制执行时间
Flink streaming是一个流式计算框架，它可以处理无界数据流并提供低延迟和高吞吐量的实时数据处理能力。在Flink streaming中，可以通过以下几种方式来控制执行时间：
- 时间窗口（Time Windows）：Flink streaming支持基于时间的窗口操作，可以将数据流划分为固定长度的时间窗口，然后对每个窗口中的数据进行处理。常见的时间窗口包括滚动窗口（Tumbling Windows）、滑动窗口（Sliding Windows）和会话窗口（Session Windows）。通过设置窗口的大小和滑动步长，可以控制窗口的执行时间。
- 水位线（Watermarks）：Flink streaming使用水位线来处理事件时间（Event Time）语义。水位线是一种衡量事件时间进展的机制，可以用来控制窗口的触发时间。通过设置水位线的值，可以告诉Flink何时认为数据流中的事件已经全部到达，从而触发窗口计算。
- 触发器（Triggers）：Flink streaming提供了多种触发器来控制窗口的计算和输出时间。常见的触发器包括基于元素数量的触发器和基于处理时间（Processing Time）的触发器。通过设置触发器的条件和策略，可以控制窗口的执行时间和输出时间。
- 延迟容忍（Latency Tolerance）：Flink streaming具有良好的延迟容忍性，可以在保证数据处理的实时性的同时，容忍一定的延迟。通过调整延迟容忍的参数，可以控制Flink streaming的执行时间和延迟。
总结起来，Flink streaming可以通过时间窗口、水位线、触发器和延迟容忍等机制来控制执行时间，从而实现对实时数据流的灵活处理。在腾讯云上，可以使用腾讯云流计算Oceanus（https://cloud.tencent.com/product/oceanus）来进行Flink streaming的开发和部署。
相关·内容
基于Flink Scala快速入门，我已经创建了一个使用org.apache.flink.streaming.connectors.twitter.TwitterSource的示例作业。
我使用本地流作业管理器/start-local-streaming.sh，并使用~/flink-0.10.0/bin/flink run target/quickstart-0.1.jar启动作业
在TwitterSource源代码中，我注意到有INFO级别的日志记录语句。如何将这些日志记录到控制台？
浏览 0提问于2015-12-04得票数 2
2回答
只是尝试从flink 1.3迁移到1.4，并在linux机器上遇到此异常：(不能在windows上重现)。
我还导入了这个包：
// https://mvnrepository.com/artifact/org.apache.flink/flink-shaded-hadoop2
compile group: 'org.apache.flink', name: 'flink-shaded-hadoop2', version: '1.4.0'
有什么帮助吗？
在flink控制台：
TriggerWindow(TumblingProcessingTim
浏览 22提问于2018-02-13得票数 1
我已经编写了一个消费者，它读取卡夫卡主题，并使用StreamSink以拼图格式写入数据。但是我得到了以下错误
java.lang.Exception: org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException: Could not forward element to next operator
at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.checkThrowS
浏览 6提问于2020-02-29得票数 0
我目前正在使用Flink 1.7 +gcs-连接器库。我试图让StreamingFileSink写到GCS桶中，并遇到以下异常：
我遇到了一个Jira： --但我不清楚代码是否曾经合并过。
在需要做什么方面有什么帮助是非常感谢的，以使这一工作？
java.lang.UnsupportedOperationException: Recoverable writers on Hadoop are only supported for HDFS and for Hadoop version 2.7 or newer
at org.apache.flink.runtime.fs.hdfs.Ha
浏览 3提问于2019-12-03得票数 1
回答已采纳
当StreamingFileSink尝试重命名进程中的文件时，我偶尔会遇到一个问题。请参见下面的异常：
文件系统挂载在Google Cloud Storage buckets上。它在大多数情况下都工作得很好，但突然失败了。
我已经验证了inprogress文件是否存在。任何提示和建议都将非常有用--谢谢！
2019-11-12 01:26:44.908 [flink-akka.actor.default-dispatcher-77] INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Job kafka-ingestio
浏览 25提问于2019-11-12得票数 0
我刚开始学习Flink中的窗口函数。我有一个定制的源码来产生数字，我的目标是计算偶数和奇数的和。
下面是代码(Flink:1.12 Scala:2.11.8)。
object ProcessWindowFunc {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
val source = env.addSource(new CustomSource())
source
.keyBy(x=>x
浏览 36提问于2021-03-03得票数 0
下面是Flink关于如何将WatermarkStrategy与KafkaConsumer结合使用的文档。代码如下所示
val kafkaSource = new FlinkKafkaConsumer[MyType]("myTopic", schema, props)
kafkaSource.assignTimestampsAndWatermarks(
WatermarkStrategy
.forBoundedOutOfOrderness(Duration.ofSeconds(20)))
val stream: DataStream[MyType] = env.ad
浏览 5提问于2020-11-23得票数 4
回答已采纳
当我执行我的Flink应用程序时，它给出了这个NullPointerException：
2017-08-08 13:21:57,690 INFO com.datastax.driver.core.Cluster - New Cassandra host /127.0.0.1:9042 added
2017-08-08 13:22:02,427 INFO org.apache.flink.runtime.taskmanager.Task - TriggerWindow(TumblingEventTimeWindows(30000), List
浏览 5提问于2017-08-08得票数 8
回答已采纳
我有一个简单的代码，如下所示，当我在IDE中运行它时，控制台上没有打印任何东西，有人可以帮我看看吗？谢谢
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.stre
浏览 32提问于2019-07-20得票数 0
回答已采纳
我正在使用Flink 1.13.1，并试图用10k记录的RPS(每秒速率)将数据写入kafka。我有一个由30个经纪人组成的kafka集群，我的flink工作做了一个过滤器操作员，只将数据汇到kafka。下面是我的生产者设置，最初我有5个接收器主题分区，但现在10个，我仍然得到同样的问题。另外，我尝试将request.timeout.ms设置为1分钟，而不是卡夫卡默认的30秒，但仍然得到了120001 ms has passed since batch creation。由于此错误，检查点失败。检查点大小仅为107 is，因为flink只向kafka提交偏移。作业并行性为36。这是纳卡的属性和
浏览 23提问于2022-04-11得票数 1
1回答
我有一个在生产环境中运行的Flink流作业，我需要对主转换代码进行更改。
生产中的代码实际上如下所示：
stream
.filter(inboundData -> inboundData.hasToBeFiltered())
.uid("filtered-data")
.keyBy(data -> data.getMyStringKey())
.process(doSomething())
.uid("processed-inbound-data-id");
我需要改变keyBy()操作符使用inboundData PO
浏览 0提问于2019-08-08得票数 1
我有一个scala文件，包含这个导入和这个代码。我不知道如何解决这个重载构造函数的问题：
package uimp
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.util.serialization.
{SimpleStringSchema,JSONKeyValueDeserializationSchema}
import org.apache.flink.core._
浏览 4提问于2020-06-02得票数 0
我正在尝试使用Scala和Flink将消息发布到Kafka主题中。但是，当使用中提供的代码创建FlinkKafkaProducer对象时，它告诉我无法应用构造器。以下是代码示例：
val studentProducer = new FlinkKafkaProducer[String](
"my_topic", // target topic
new SimpleStringSchema(), // serialization schema
properties, // producer config
浏览 103提问于2021-05-24得票数 1
需要帮助！谁能给我指引一条正确的路？
以下是我的代码和日志的片段。
DataStream<ObjectNode> stream = env.addSource(KafkaConsumer.getKafkaConsumer());
DataStream<MyDataObject> dataStream = stream.flatMap(new DataTransformation());
我使用flatMapFunction来处理我的输入对象并获得多个对象。
以下是stackTrace：
java.lang.RuntimeException: Buffer pool is
浏览 8提问于2018-01-16得票数 4
我正在尝试向我的flink集群提交一个作业，但我一直遇到以下错误：
2021-05-03 17:14:32
java.lang.NoSuchMethodError: org/apache/flink/api/common/state/OperatorStateStore.getSerializableListState(Ljava/lang/String;)Lorg/apache/flink/api/common/state/ListState; (loaded from file:/opt/flink/lib/flink-dist_2.11-1.11.3.jar by jdk.interna
浏览 5提问于2021-05-04得票数 1
我们正在使用包括map、windowing、flatMap()在内的几个运算符运行一个Flink作业，作业失败并出现以下错误-只是想知道是什么导致了这个错误：
2021-05-27 07:00:07,023 WARN org.apache.flink.runtime.taskmanager.Task [] - Collect Inventory (1/1)#0 (34d81cf2e59f350886f93a1e0f734d38) switched from RUNNING to FAILED with failure cause: org.apache.
浏览 229提问于2021-05-27得票数 0
这一次我的问题是:使用MapState，使用mapstate.put(key, value)将键的当前值修改到mapState中是安全的，或者我需要执行mapState.remove(key)，然后再次执行mapstate.put(key, value)，或者无论如何都要更新此值？ 从Flink的状态抽象开始，抽象不是为并发访问而设计的，也不应该在多个线程之间共享。那么，重新表述我的问题:我可以根据键将值更新到mapState中，而不需要删除键然后再放入键吗？如何避免使用mapState的ConcurrentModificationException，而不将该操作符的并行度设置为1？ 因为我
浏览 43提问于2020-04-16得票数 1
我遵循的是，但是构建了Flink 1.5-SNAPSHOT。
我做错了什么？
[INFO] ------------------------------------------------------------------------
[INFO] Building flink-streaming-java 1.5-SNAPSHOT
[INFO] ------------------------------------------------------------------------
[INFO]
[INFO] --- maven-clean-plugin:3.0.0:clean (
浏览 4提问于2018-01-19得票数 1
回答已采纳
1回答
2022-01-04 15:58:17 java.lang.NoClassDefFoundError:无法初始化org.apache.flink.streaming.connectors.elasticsearch7.Elasticsearch7ApiCallBridge.createBulkProcessorBuilder(Elasticsearch7ApiCallBridge.java:80) at org.apache.flink.streaming.connectors.elasticsearch7.Elasticsearch7ApiCallBridge.createBulkProce
浏览 6提问于2022-01-04得票数 0
嗨，我用下面的代码发布从flink到Kafka的消息。
final Properties props = new Properties();
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALU
浏览 14提问于2022-06-14得票数 0
回答已采纳
作为标题，异常发生在键控窗口中，
java.lang.IllegalArgumentException: Key Group 91 does not belong to the local range.
at org.apache.flink.util.Preconditions.checkArgument(Preconditions.java:139)
at org.apache.flink.streaming.api.operators.HeapInternalTimerService.getIndexForKeyGroup(HeapInternalTimerService.java:43
浏览 2提问于2018-07-17得票数 1
1回答
我正在尝试运行一个简单的程序，它从一个动态流中读取，做一个琐碎的转换，并将结果写入另一个动态流。
在Flink 1.4.0上本地运行(这是目前EMR支持的版本，因此无法升级)。
以下是代码：
def main(args: Array[String]) {
val env = StreamExecutionEnvironment.getExecutionEnvironment
val consumerConfig = new Properties()
consumerConfig.put(AWSConfigConstants.AWS_REGION, "us-east-1&#
浏览 1提问于2018-06-13得票数 0
回答已采纳
1回答
我在flink上有个问题。我实时计算引擎使用groovy脚本扩展计算类型(如:sum、average、、count等)。我们定义了一个标准的计算接口(AbstractCompute)，如果我想在这个框架中扩展一个计算类型，我只需要执行AbstractCompute.and，然后将groovy脚本存储在数据库中。然后，应用程序可以逐个任务地读取脚本，并通过GroovyClassLoader加载到虚拟机中。这个程序不用再用Flink了，看工作做的很好。原因是Flink使用另一个ClassLoader (FlinkUserCodeClassLoaders$ChildFirstClassLoader)
浏览 2提问于2019-01-14得票数 2
0回答
所以我用它安装了一个样例Flink项目。
我正在尝试使用这个模板，它让我可以开始编写一个Flink摄取应用程序，而不必担心依赖关系，但它适得其反。当我尝试与sbt同步时，我发现找不到Flink (不是要说谎，但这部分对我来说甚至是模糊的)。我想知道是否有人知道如何让我的项目找到Flink。使用这项技术非常令人兴奋。
Error while importing sbt project:
Java HotSpot(TM) 64-Bit Server VM warning: ignoring option MaxPermSize=384M; support was removed in 8.0
[
浏览 6提问于2017-12-05得票数 0
回答已采纳
我正在开发一个沉入Kafka的Flink应用程序。我创建了一个默认池大小为5的Kafka生产者。我使用以下配置启用了检查点：
env.enableCheckpointing(1800000);//checkpointing for every 30 minutes.
// set mode to exactly-once (this is the default)
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// make sure 500 m
浏览 7提问于2020-03-21得票数 1
我正在试用Flink的新Python streaming，并尝试用./flink-1.6.1/bin/pyflink-stream.sh examples/read_from_kafka.py运行我的脚本。python脚本相当简单，我只是尝试使用现有主题中的内容，并将所有内容发送到标准输出(或日志目录中的*.out文件，默认情况下，输出方法将在该文件中发出数据)。
import glob
import os
import sys
from java.util import Properties
from org.apache.flink.streaming.api.functions.sour
浏览 0提问于2018-10-10得票数 4
我运行了一个flink作业，18小时后它失败了。失败消息:检查点协调器正在挂起。
检查点屏幕截图：
工作概述屏幕截图：
以下是作业管理器日志：
2020-10-10 13:53:10,636 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph - Source: kafkaSource -> Timestamps/Watermarks -> Process (1/1) (c38419ece8208c1ef2948087f2b84dd0) switched from RUNNING to
浏览 3提问于2020-10-10得票数 4
我使用flink监控hdfs中的新文件(文件为gzip格式)，并对其进行处理。 env.readFile(filePath) 它可以在文件有效时工作， 但如果gzip文件无效，flink作业将被终止。 有异常日志： java.io.IOException: Error opening the Input Split hdfs://mdw:8020/user/data/15_077_4.gz [0,-1]: Not in GZIP format
at org.apache.flink.api.common.io.FileInputFormat.open(FileInputFormat.
浏览 77提问于2021-04-17得票数 0
回答已采纳
我正在对以下代码进行一些查询：
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
DataStream<Row> ds = SourceHelp.builder().env(env).consumer010(MyKafka.builder().build().kafkaWithWaterMark2())
.rowTypeInfo(MyRowType.builder().build().typeInfo())
.build().source4()
浏览 1提问于2018-09-17得票数 1
我想做一个流媒体作业。
当我尝试使用start-clusted.sh和Flink在本地运行它时，我没有问题。
但是，我目前正在尝试使用Flink on try (部署在Google上)来运行我的工作，当我试图取消它时，取消状态将永远持续，并且TaskManager中仍有一个插槽。
这是我得到的日志：
2016-10-18 16:56:04,053 INFO org.apache.flink.runtime.taskmanager.Task -
Attempting to cancel task Source: pubSubMessageAcknowledgingSource ->
T
浏览 4提问于2016-10-19得票数 2
回答已采纳
下面提到的Apache测试代码在我想使用KeyedOneInputStreamOperatorTestHarnessK、IN、Int或Long作为key (K)参数时不会在Scala中编译。可以使用字符串、整数或java.lang.Long类型的泛型参数编译相同的代码：
import org.apache.flink.api.java.functions.KeySelector
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.functions.KeyedProces
浏览 7提问于2021-12-31得票数 2
我正在创建一个测试，以查看我的flink模式的超时是否正确。为此，我使用了flink spector，我有以下测试案例：
@Test
public void SameDoor_TwoStatuses_OneSecondTimeoutPattern() {
// Arrange
long now = new Date().getTime();
DoorEvent event1 = new DoorEvent();
event1.setId(123);
event1.getDoor().setId(1);
event1.getDoor().setSt
浏览 1提问于2017-09-15得票数 0
回答已采纳
我正在尝试运行流作业，它消耗来自Kafka的消息，转换它们，并将其下沉到Cassandra。
当前代码片段失败
val env: StreamExecutionEnvironment = getExecutionEnv("dev")
env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime)
.
.
.
.
val source = env.addSource(kafkaConsumer)
.uid("kafkaSource")
浏览 2提问于2020-05-27得票数 1
我的程序想在没有Flink窗口的情况下加入两个流。
我连接了两个流并定义了一个类A扩展了RichCoFlatMapFunction来处理它们。然后Guava缓存有一个删除侦听器来收集加入和过期的数据到下一个Flink函数。
private synchronized void collect(ReqFeatures features) {
feaCollector.collect(features);
}
每次开始时，它都运行得很好，但几个小时后，它总是因为这个异常而死。
java.io.IOException: Corrupt stream, found tag: 105
at org
浏览 4提问于2019-03-07得票数 0
1回答
我有一个简单的模式，如下所示
Pattern<Event,?> pattern = Pattern.<Event>begin("s1")
.where(new SimpleCondition<Event>() {
@Override
public boolean filter(Event event) throws Exception {
Long
浏览 0提问于2018-01-04得票数 0
我的Flink管道目前使用一个Pojo，其中包含一些列表和映射( Strings)，按照
public class MyPojo {
private List<String> myList = new ArrayList<>();
private OtherPojo otherPojo = new OtherPojo();
// getters + setters...
}
public class OtherPojo {
private Map<String, String> myMap = new HashMap<
浏览 16提问于2020-01-18得票数 3
回答已采纳
1回答
我有一份Flink的工作，把数据写进卡夫卡。Kafka主题的最大消息大小设置为5MB，所以如果我尝试编写任何大于5MB的记录，它会抛出以下异常并将作业降低。
java.lang.Exception: Failed to send data to Kafka: The request included a message larger than the max message size the server will accept.
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase.checkErroneo
浏览 1提问于2018-09-13得票数 1
回答已采纳
1回答
我试着用阿帕奇光束和Flink在AWS KDA。这个管道读取一些动态数据，经过简单的转换后，它尝试使用以下策略将结果分组为一个碎片：
.apply("Creating micro-batches", Window.<XXX>into(new GlobalWindows())
.triggering(Repeatedly.forever(AfterFirst.of(AfterPane.elementCountAtLeast(windowSizePanes),
浏览 3提问于2021-12-20得票数 0
回答已采纳
我刚开始使用flink，我正在实现一个模式识别模块(不使用CEP实现模式匹配)，它将从EventHub主题读取json流，如果模式匹配，则推送到另一个事件中心主题。我的模块功能如下
从接收JSON负载
I使用的是一个RichSourceFunction，它将从API读取模式并发送到广播流
我正在使用Flink BroadcastProcessFunction来根据广播状态中可用的模式列表来处理数据，并且我没有使用键流或我的程序中的任何状态，因为没有状态的范围，我只需要检查某些值是否存在于JSON .中。
我读到flink在内部维护状态，即使没有状态显式实现
下面的设置为检查点，我平均每小时收到
浏览 8提问于2021-01-29得票数 0
在将消息发布到flink应用程序的kafka主题时，会出现此错误。相同的代码在我们的测试环境中工作，具有类似的生产者配置，但在生产环境中失败。我找不到这个问题的原因。
2022-04-15 16:51:37,228 thread="Sink: spend-limit-publisher-sink (2/2)#15" level=WARN logger=o.a.f.r.t.Task - Sink: spend-limit-publisher-sink (2/2)#15 (d7b4646c840c3882bb784125393b484a) switched from RUNNIN
浏览 1提问于2022-04-15得票数 0
Added Depedency Pom Details :
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.11</artifactId>
<version>1.7.1</version>
</dependency>
<depe
浏览 0提问于2019-04-04得票数 2
回答已采纳
零星的异常发生如下。
java.lang.RuntimeException: Error triggering a checkpoint as the result of receiving checkpoint barrier
at org.apache.flink.streaming.runtime.tasks.StreamTask$2.onEvent(StreamTask.java:701)
at org.apache.flink.streaming.runtime.tasks.StreamTask$2.onEvent(StreamTask.java:691)
a
浏览 3提问于2016-06-27得票数 0
我试图使用flink 1.15.1编写一个简单的表API S3流接收器(csv格式)，并面临以下例外：
Caused by: org.apache.flink.util.SerializedThrowable: S3RecoverableFsDataOutputStream cannot sync state to S3. Use persist() to create a persistent recoverable intermediate point.
at org.apache.flink.fs.s3.common.utils.RefCountedBufferingFileStream
浏览 20提问于2022-07-11得票数 0
回答已采纳
1回答
我正在尝试打印一个字符串，如果使用Flink CEP库找到Hello和world。我的来源是Kafka，并使用控制台生产者输入数据。这一部分正在发挥作用。我可以打印出我在主题中输入的内容。但是，它不会打印出我的最后一条消息“世界真好！”。它甚至不会打印出它进入了lambda。下面是类
package kafka;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.fl
浏览 10提问于2016-09-19得票数 2
1回答
在我的Flink反序列化程序中，我从getProducedType方法中得到以下错误：
com.esotericsoftware.kryo.KryoException: java.lang.NullPointerException
Serialization trace:
values (org.apache.avro.generic.GenericData$Record)
at com.esotericsoftware.kryo.serializers.ObjectField.read(ObjectField.java:125)
at com.esotericsoftware
浏览 31提问于2018-01-03得票数 3
回答已采纳
我试图在flink服务器上运行一些代码，但无法
下面是我的代码
DataStream<UserInfo> keyedStream = executionEnvironment
.addSource(new UserDataSource());
keyedStream.assignTimestampsAndWatermarks(new MessageWaterEmitter());
tableEnv.registerDataStream("test", keyedStream, "us
浏览 0提问于2020-12-30得票数 0
回答已采纳
当我试图用flatMap操作符写出集合时，我得到了非法的状态异常(仅在高负载下)：缓冲池被破坏了，我在这里做错了什么？当flink抛出缓冲池错误时？
java.lang.RuntimeException: Buffer pool is destroyed.
at org.apache.flink.streaming.runtime.io.RecordWriterOutput.pushToRecordWriter(RecordWriterOutput.java:110)
at org.apache.flink.streaming.runtime.io.RecordWriter
浏览 1提问于2019-04-02得票数 3
回答已采纳
我有一个基于flink的流媒体应用程序，它使用apache kafka源和接收器。因为有时候我会在开发过程中的随机时间得到异常，我不知道它们是从哪里来的。 我使用mainRunner类在IntelliJ中运行应用程序，并通过kafka向它提供消息。有时第一条消息会触发错误，有时只有在几条消息之后才会发生。 它看起来是这样的： 16:31:01.935 ERROR o.a.k.c.producer.KafkaProducer - Interrupted while joining ioThread
java.lang.InterruptedException: null
at
浏览 111提问于2019-01-29得票数 1
回答已采纳
3回答
我正在尝试阅读卡夫卡主题的数据，在Flink流中。我正在尝试运行下面的示例代码，作为APACHE 1.1.3文档中的示例: Apache连接器，
import java.util.Properties;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09;
import org.apache.flink.streaming.api.datastrea
浏览 4提问于2017-01-09得票数 2
回答已采纳
我们有一个flink流作业，它读取卡夫卡的数据并将其汇到S3。我们使用flink的内部流文件接收器API来实现这一点。然而，几天后，作业失败了，无法从失败中恢复。消息说它无法从找到文件。我们想知道可能的根本原因是什么，因为我们真的不想丢失任何数据。
谢谢。
整个输出如下所示
java.io.FileNotFoundException: No such file or directory: s3://bucket_name/_part-0-282_tmp_b9777494-d73b-4141-a4cf-b8912019160e
at org.apache.flink.fs.shaded.
浏览 1提问于2019-11-27得票数 1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
