flink流是不变的吗？
Flink流是不变的。Flink是一个开源的流处理框架，它支持在分布式环境中进行实时数据流处理和批处理。Flink的流数据模型是基于事件时间的，它将数据流看作是一系列有序的事件，这些事件可以是无限的，也可以是有限的。
Flink流的不变性体现在以下几个方面：
- 事件顺序：Flink保证事件在流中的顺序不变。无论是在单个任务中还是在分布式集群中，Flink都会按照事件的产生顺序对其进行处理，确保事件的顺序性。
- 状态一致性：Flink使用状态来存储和管理流处理过程中的中间结果和状态信息。Flink保证在处理过程中，状态的一致性和正确性。即使在发生故障或重启的情况下，Flink也能够恢复到之前的状态。
- 容错性：Flink具有高度的容错性，能够应对节点故障和网络异常等情况。它通过将数据流划分为小的、可重放的数据块，并将这些数据块进行分布式备份，以保证数据的可靠性和一致性。
- 事件时间处理：Flink支持基于事件时间的处理，即按照事件发生的时间顺序进行处理。这种处理方式可以解决乱序事件和延迟事件的问题，保证结果的准确性。
- 状态更新：Flink中的状态是可变的，可以根据业务需求进行更新。通过状态的更新，Flink可以实现更复杂的流处理逻辑。
Flink流的应用场景非常广泛，包括实时数据分析、实时报表生成、实时推荐系统、欺诈检测、网络监控等。对于Flink流的处理，腾讯云提供了相应的产品和服务，如腾讯云流计算Oceanus，详情请参考：https://cloud.tencent.com/product/oceanus
相关·内容
我刚刚开始学习apache，并遇到了一个问题:我怎样才能暂停flink工作，然后继续它呢？flink支持使用命令行挂起作业吗？
浏览 3提问于2017-06-13得票数 1
为什么Flink-HBase connector中没有HBaseReadStreamExample 似乎有一个读、写和WriteStream示例，但没有ReadStream。 从HBase到Flink的流读取是一种糟糕的做法吗？
浏览 21提问于2020-07-03得票数 1
回答已采纳
Spark有mapPartition API，而Flink DataStream API没有。有谁能帮助解释原因吗？我想要做的是在Flink上实现一个类似于星火reduceByKey的API。
浏览 2提问于2015-10-28得票数 7
回答已采纳
我试图用Flink 1.5.0创建一个面向批处理的Flink作业，并希望使用Table和SQL来处理数据。我的问题是试图创建一个编译错误的BatchTableEnviroment
org.apache.flink.streaming.api.environment.StreamExecutionEnvironment无法访问BatchJob.java:46,73
引起于
final BatchTableEnvironment bTableEnv = TableEnvironment.getTableEnvironment(bEnv);
据我所知，我不依赖于流环境。我的代码是下面的代码片
浏览 1提问于2018-06-19得票数 0
我正在开发一个POC，将现有的Flink应用程序/拓扑转换为使用KafkaStreams。我的问题是关于部署。
具体来说，在Flink中，可以向flink安装添加"Worker节点“，然后为拓扑添加更多的并行化，以跟上不断增长的数据速率。
如何随着数据速率的增加而增加KStreams容量？KStreams会自动处理这个问题吗？我是否启动更多的进程(ala微服务)？
还是我错过了这里的大局？
浏览 4提问于2016-08-17得票数 3
回答已采纳
正在尝试在Amazon EMR Flink(5.21.0)上运行Flink(v1.7.0)作业。我得到了异常 java.lang.NoSuchMethodError: scala.Product.$init$(Lscala/Product;)V 异常看起来像是SCALA版本问题。我发现flink库是SCALA 2.11，而我的工作是使用2.12构建的。Flink 1.7.0支持SCALA 2.12。 问题是，如何让Amazon EMR支持SCALA 2.12而不是2.11？或者，除了我的代码回到2.11之外，还有更好的解决方案吗？
浏览 12提问于2019-03-01得票数 0
我是新的flink用户，我有以下问题。使用flink将从关系数据库中提取的相关数据传输到HBase中。我用多个ExecutionEnvironments (每个RDB表一个用于并行传输表行)在java上编写flink批处理应用程序，以便按表顺序地传输表(因为调用env.execute()是阻塞的)。
我像这样开始纱线会话
export YARN_CONF_DIR=/etc/hadoop/conf
export FLINK_HOME=/opt/flink-1.3.1
export FLINK_CONF_DIR=$FLINK_HOME/conf
$FLINK_HOME/bin/yarn-sessi
浏览 1提问于2018-01-16得票数 7
回答已采纳
我试图在flink服务器上运行一些代码，但无法
下面是我的代码
DataStream<UserInfo> keyedStream = executionEnvironment
.addSource(new UserDataSource());
keyedStream.assignTimestampsAndWatermarks(new MessageWaterEmitter());
tableEnv.registerDataStream("test", keyedStream, "us
浏览 0提问于2020-12-30得票数 0
回答已采纳
我有两个来自卡夫卡的事件源，一个来自user主题，另一个来自order主题，这两个主题的事件在我的代码中被创建为Flink动态表。
我有以下简单的连接查询，希望计算每个用户所放置的订单数。
select user_id,count(order_id) from user join order on user.user_id = order.user_id，
我有以下问题：
假设应用程序开始运行时，用户和订单中没有数据。
当Flink连接器从Kafka主题获得新数据时，会立即运行查询任务吗？查询中将包括多少个事件？我一次从卡夫卡控制台写了1000个用户和订单。。
如果查询(连接查询或复杂查询)需
浏览 6提问于2022-07-24得票数 0
回答已采纳
我使用的是运行在Docker上的Flink 1.2.1，任务管理器作为Docker群的一部分分布在不同的am上。
使用Flink Web UI上传Apache光束应用程序并尝试在作业提交点设置并行度不起作用。两者都不会使用Flink CLI提交作业。
看起来在客户机级并行度并没有被接受，它最终默认为1。
当我以编程方式在Apache Beam代码中设置并行度时，它可以工作：flinkPipelineOptions.setParallelism(4);
我怀疑问题的根源可能在org.apache.beam.runners.flink.DefaultParallelismFactory类中，因为它
浏览 2提问于2017-10-23得票数 0
Flink是流媒体的一个明显的选择，然而Spark已经成熟了ML pipelines，它可以在spark中训练模型，将其保存为CrossValidatorModel并部署到Flink中的流媒体数据吗？全部在scala中
浏览 16提问于2017-03-16得票数 0
回答已采纳
1回答
我正在尝试将Flink stateful函数部署为flink jar，并遵循了here的说明。然而，我得到的错误是，即使我在pom中添加了依赖项，程序入口点类也没有在jar中找到。 运行： bin/flink run -c org.apache.flink.statefun.flink.core.StatefulFunctionsJob flink-statefun/statefun-examples/statefun-greeter-example/target/statefun-greeter-example-2.3-SNAPSHOT.jar 产生以下错误： org.apache.fli
浏览 237提问于2020-12-23得票数 2
我试图使用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
回答已采纳
我正在读
，
它有以下例子：
SELECT *
FROM Orders o, Shipments s
WHERE o.id = s.order_id
AND o.order_time BETWEEN s.ship_time - INTERVAL '4' HOUR AND s.ship_time
我有两个问题：
如果o.order_time和s.ship_time是正常的时间列，而不是事件时间属性，那么所有的状态都会保存在Flink中，就像普通的规则内部连接那样？这样，也许大尺寸的状态就会保持在Flink中。
如果o.order_time和s.ship_time是事件时间属性，那
浏览 0提问于2021-08-06得票数 0
回答已采纳
1回答
用于业务逻辑类的Apache
、、、、
我正在使用Apache和Kakfa，在这里我能够正确地使用来自主题的消息并反序列化它。
我正在使用flink datastream Map函数来处理收到的kafka消息，但是flink期望我的所有业务对象都是可序列化的。
下面是伪代码
TestRequestListener listener = this.context.getBean(TestRequestListener.class);
KafkaConsumer<KafkaTestBatch> kafkaConsumer = KafkaSource<>.builder()...build();
StreamExe
浏览 2提问于2022-01-30得票数 0
我有一个像这样的流：<_time(timestamp), uri(string), userId(int)>。_time属性是rowtime，我将它注册为一个表：
tableEnv.registerDataStream("userVisitPage", stream, "_time.rowtime, uri,userId");
然后我查询表：
final String sql =
"SELECT tumble_start(_time, interval '10' second) as timestart,
浏览 1提问于2018-03-02得票数 1
1回答
为了丰富数据流，我们计划将MySQL (MemSQL)服务器连接到现有的flink流应用程序
正如我们所看到的，Flink提供了一个带有JDBC连接器https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/table/jdbc/的Table
此外，我还发现了另一个名为Flink-CDC https://ververica.github.io/flink-cdc-connectors/master/content/about.html的MySQL连接器，允许以流的方式处理外部数据库。
他们之间有什么区别？在我的情况
浏览 19提问于2022-02-07得票数 0
回答已采纳
1回答
我对flink很陌生，并试图将其部署到EMR集群上。我使用了3个节点集群(1个主服务器和2个从节点)进行默认配置。我没有做任何配置更改，并坚持默认配置。
我很想了解以下几点：
主从是如何相互通信的，因为我还没有在主节点中提到任何IP？
我可以在主节点(Path: /usr/lib/ flink )中看到flink库，但在从节点中找不到flink库。我的代码如何在从节点上执行？
如果需要，我将根据conf/flink-config.yml中的要求更改一些配置。除了这个之外，我还需要在主节点或从节点上做任何其他更改吗？
浏览 1提问于2018-05-10得票数 3
我计划动态缩放一个Flink应用程序。该应用程序使用kafka-flink连接器消耗来自Kafka的事件。
因为应用程序的“热身”只需几分钟(缓存.)而改变并行级别涉及重新启动，我更喜欢提交(放大)或交替杀死(缩小)任务，而不是改变并行级别。
我想知道，从性能、逻辑和执行计划来看，这种方法和Flink内置的并行执行有什么不同吗？
换句话说，10个相同的Flink任务与一个并行级别= 10的任务( env.setParallelism( 10 ) )有什么区别？
浏览 1提问于2019-09-25得票数 0
回答已采纳
1回答
我有一个定制的flink Source，还有一个SerializableTimestampAssigner，它将事件时间戳分配给源发出的记录。由于底层数据存储的性质，源可能会发出无序的记录，但是在批处理模式下，我希望在任何操作符处理这些记录之前，Flink会根据事件时间戳对这些记录进行排序。
摘自Flink 的执行模式：
在批处理模式中，在预先知道输入数据集的情况下，不需要这样的启发式，因为至少可以按时间戳对元素进行排序，以便按时间顺序处理元素。
然而，情况似乎并非如此。如果我用时间戳分配器在Source (StreamExecutionEnvironment.fromSource)中
浏览 1提问于2022-04-22得票数 1
回答已采纳
我需要在K8上部署Flink应用程序的帮助 我们有3个源，将以SQL查询的形式发送触发条件。总查询数约为3-6k，有效地增加了flink实例的负载。我试着执行，但它非常慢，并且需要很多时间才能开始。 由于大量的查询，我们决定为每个源创建多个flink应用实例。因此，一个flink实例实际上只能执行~1-2K个查询。 示例: sql查询源为A、B、C Flink实例： App A -->将只负责处理源A查询 App B -->将只负责处理源B查询 App C -->将只负责处理源代码C查询 我想在Kubernetes上部署这些实例 问题： a)可以使用迷你集群(内置)部署独立
浏览 31提问于2020-08-24得票数 1
我使用intellij、maven 3和flink 1.15.1编写有状态流作业。我正在尝试为我的自定义KeyedProcessFunction编写单元测试，并尝试遵循文档以及添加提到的的依赖项。我对使用KeyedOneInputStreamOperatorTestHarness类很感兴趣，但是在我的任何依赖项中都找不到这个类(下面发布的修改后的pom )。我唯一能找到这门课的方法就是
<dependency>
<groupId>org.apache.flink</groupId>
<ar
浏览 8提问于2022-08-01得票数 1
回答已采纳
2回答
如何在Flink转换中更新配置？
、、、、
给定一个Flink流作业，该作业将map()操作应用于流。
此map()操作从某些属性读取其配置，并相应地映射数据。例如，配置指定读取属性"input"，并使用不同的属性名"output“将其写入流。这个已经很好了。
现在，配置发生了变化，例如，转换将为输出使用不同的属性名。
因此，我正在寻找一种方式，让所有的Flink任务重新阅读一个新的配置在运行时。
有没有可能
将KafkaSourcewait挂起，直到管道耗尽(flush)trigger集群中的所有任务，以重新读取配置文件(coordinated)resume (coordinated)resume KafkaS
浏览 13提问于2022-10-11得票数 0
我想将流SQL放在Kafka中，以供Flink for CEP使用。这是一种好方法吗？
我知道动态模式定义在Flink上是不允许的，我需要应用它们可以在无界事件流上更改的规则。
举个例子；
有一个供用户定义其设备规则的UI。想象一下，有一个事物跟踪服务(如亚马逊网络服务中心)保持物理设备的状态，我也认为这是将每个设备的特定规则放入影子规则中的好方法，当影子参与者接收到传感器数据时，它可以发出数据，并将其规则添加到IoT作为规则引擎使用(通过kafka)。因此，我希望Flink执行传入的传感器数据(及其规则)，这些数据对于每个设备都可能是不同的。
浏览 7提问于2017-07-20得票数 1
据我理解，Flink中的运算符有源运算符、变换运算符等。对于Flink中的运算符，我的理解是否正确？
在操作符状态中，Flink是为每个作业/任务维护每个操作符的状态(map()、or ()等)还是维护一个完整作业/任务的状态？另外，如果我的作业提交了多个并行性，那么每个槽会有自己的状态吗？
假设，我提交了两个作业，它们是键流，两个作业都有相同的键，比如“颜色”，但这两个任务完全不同。Flink是否会维持两个不同的州，或者它将维持两个工作的一个州。
浏览 0提问于2019-07-22得票数 2
回答已采纳
我正在为一个大容量流用例(每秒数以千万计的事件)开发一个DataStream-based Flink应用程序。这些数据是从卡夫卡主题中消耗的，并且已经根据特定的键进行了切分。我的意图是在Flink端创建特定于键的状态，以运行自定义分析。我无法解决的主要问题是，如何创建键控状态，而不对keyBy()强加的传入数据进行重新配置。
我可以保证Flink作业的最大并行性将小于或等于源Kafka主题中的分区数，因此逻辑上没有必要进行洗牌。 StackOverflow问题的答案表明，可能有可能以一种符合Flink预期的方式将数据写入Kafka，然后使用reinterpretAsKeyedStream()。
浏览 4提问于2022-08-08得票数 2
回答已采纳
比较Flink和Spark Streaming，Flink和Storm，Storm和Heron有很多问题。
这个问题的来源是Apache Flink和Twitter Heron都是真正的流处理框架(而不是像Spark Streaming这样的微批处理框架)。Storm已于去年被Twitter淘汰，他们改用Heron (基本上是Storm重新制作的版本)。
在Flink和Flink vs Spark上有斯利姆·巴尔塔吉的精彩演讲：
Ilya Ganelin对各种流媒体框架的出色研究：
关于Flink vs Storm的有趣想法：
但我还没有看到新的Storm/Heron与Apache Flink
浏览 27提问于2016-06-05得票数 9
我有两个流。第一个是基于时间的流，我使用countTimeWindow接收前10个数据点来计算STAT值。我手动使用了变量cnt来只保留第一个窗口，并过滤了剩余的值，如下面的代码所示。 然后，我想使用这个值来过滤主流，以便获得比我在窗口流中计算的stat值更大的值。 然而，我不知道如何合并或计算这两个流来实现我的目标。 我的场景是，如果我将第一个stat值转换为广播变量，然后将其传递给主流，以便能够根据广播变量中的stat值过滤传入的值。 下面是我的代码。 import com.sun.org.apache.xpath.internal.operations.Bool;
import org
浏览 10提问于2019-02-12得票数 0
回答已采纳
2回答
我在流的世界里是个新手，在我的第一次尝试中我遇到了一些问题。 更具体地说，我正在尝试使用Flink在滑动窗口中实现计数和groupBy功能。 我在一个普通的DateStream中做过，但是我不能让它在WindowedStream中工作。 你对我该怎么做有什么建议吗？ val parsedStream: DataStream[(String, Response)] = stream
.mapWith(_.decodeOption[Response])
.filter(_.isDefined)
.map { record =>
(
浏览 18提问于2019-05-02得票数 1
这段代码将pandas转换为flink表，进行转换后再转换回pandas。当我使用filter filter而不是select时，它工作得很好，但当我添加group_by和order_by时，它会给我一个错误。 import pandas as pd
import numpy as np
f_s_env = StreamExecutionEnvironment.get_execution_environment()
f_s_settings = EnvironmentSettings.new_instance().use_old_planner().in_streaming_mode().
浏览 102提问于2020-07-16得票数 0
回答已采纳
我有以下针对flink 1.12.0的简单代码，当我运行它时，异常抛出，有人能帮我看一下吗？谢谢!
package org.example.model
import java.sql.Timestamp
case class Person2(name:String, birthday: Timestamp)
和
package org.example.sql2
import java.sql.Timestamp
import java.util.Date
import org.apache.flink.streaming.api.TimeCharacteristic
import or
浏览 2提问于2020-12-25得票数 0
回答已采纳
在卡夫卡主题中，我有{“主机名”：“abc.com”、“数据”：{}}这样的事件。这些事件是json对象。我的flink-kafka消费者正在访问这些事件。现在，在flink中，我可以根据主机名对这些事件进行分组吗?在每个组中，我可以检查模式吗？
浏览 0提问于2018-12-07得票数 0
回答已采纳
我是Flink的新手，我的理解是遵循API调用
StreamExecutionEnvironment.getExecutionEnvironment().readFile(format, path)
将并行读取给定S3存储桶路径的文件。
我们将日志文件存储在S3中。要求是为多个客户端请求提供服务，以便使用时间戳从不同的文件夹读取数据。
对于我的用例，为了服务多个客户端请求，我正在评估使用Flink。因此，我希望Flink为不同的亚马逊网络服务S3文件路径并行执行亚马逊网络服务S3读取。
是否有可能在单个Flink作业中实现这一点。有什么建议吗？
浏览 49提问于2017-07-18得票数 2
我在一个关系数据库中有一些批处理的数据，我想使用Flink将它们推送到消息总线上。由于Flink同时支持批处理和流模式，因此它看起来非常适合。话虽如此，我不知道这项任务是属于StreamingJob还是BatchJob，以及如何将两者联系起来。此任务是否更适合FlinkSQL环境？ 这个是可能的吗？我需要注意什么？
浏览 25提问于2021-04-28得票数 0
回答已采纳
我想使用Apache来处理应用程序中的事件。我在独立JVM上的测试运行得相当好，尽管flink是一个非常大的依赖项。
我还试图让它在OSGi中运行，但现在放弃了，因为有许多依赖项。
所以我的问题是:我能做多小的Flink。目前，我尝试使用maven依赖于flink流-java。
不幸的是，这取决于或嵌入(只列出有问题的)：
flink-阴影-hadoop2 2
克里奥
动物园管理员
奈蒂
防波堤
apache http客户端
apache http核心
斯卡拉
阿克卡
杰克逊
它还看起来像几个jars一次又一次地嵌入相同的库。就像一些谷歌libs
浏览 4提问于2015-12-18得票数 1
我有一个带有Yarn的Flink集群，使用flink-quickstart-java原型构建一个演示项目。在使用'mvn干净包-Pbuild‘命令构建了一个胖jar之后，并使用'flink run -m丝- -yn 2./flink-快照-1.0.jar’提交程序后，程序会抛出以下异常：
org/apache/kafka/common/serialization/ByteArrayDeserializer : org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09.setDeserializer(Fli
浏览 1提问于2019-08-05得票数 3
1回答
我有一个java应用程序，它通过从数据库查询表并将其输入kafka主题来对批处理进行flink批处理。我该如何定期安排这件事。有flink调度程序吗？例如，我的java应用程序应该在后台继续运行，flink调度程序应该定期查询数据库中的表，flink批处理处理它并输入kafka(flink批处理和输入kafka已经完成了我的应用程序的一部分)。如果有人对此有指点，请帮忙。
浏览 2提问于2016-11-29得票数 4
回答已采纳
我在CheckpointedFuntion中使用运算符状态，但是在初始化MapState时遇到了NullPointerException：
public void initializeState(FunctionInitializationContext context) throws Exception {
MapStateDescriptor<Long, Long> descriptor
= new MapStateDescriptor<>(
"state",
TypeInfor
浏览 0提问于2018-03-18得票数 0
我正在为流分析方案评估Flink，并且还没有找到关于如何实现我们今天在遗留系统中所做的一种ETL设置的足够信息。
一个非常常见的场景是，我们使用了关键的、慢吞吐量的元数据流，我们希望使用这些数据流在高吞吐量数据流上进行丰富，如下所示：
这就引出了两个关于Flink的问题:如何通过缓慢更新时间窗口重叠但不相等的流来丰富快速移动的流(元数据可以存活数日，而数据可以存活几分钟)？如何有效地将多个(最多10个)数据流与Flink连接起来，比如一个数据流和9个不同的富集流？
我知道我可以使用非窗口的外部ETL缓存来实现我的ETL场景，例如使用Redis (这是我们今天使用的)，但我想看看Fli
浏览 3提问于2017-11-21得票数 1
回答已采纳
1回答
我正在研究使用Flink作为流媒体引擎。我来自apache-storm，据我所知，storm的Bolt类似于Flink的任务/操作符。在风暴中，一个人可以 builder.setBolt("TEST", new TestBolt(),5)
.fieldsGrouping("Source1", "ID1")
.fieldsGrouping("Source2","ID2)
.fieldsGrouping("Source3","ID1")
浏览 32提问于2021-07-20得票数 0
最近我阅读了flink官方文档，了解了一些关于水印的内容。 网址：https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/event_time.html 有两张图片说明了flink水印机制，这让我很困惑： flink watermark 第一幅图很容易理解，但在第二幅图中，我想知道如何获得w(11)和w(17)？我们知道，我们可以定义如何在flink作业中生成水印，换句话说，水印是按照一定的规则生成的。那么在第二张图片中生成水印的规则是什么呢？ 我查找了几乎所有不同版本flink的官方文档，它们都使用相同的图片。这让我很
浏览 28提问于2021-03-30得票数 0
image.png
没看到storm和flink的开发指南？EMR支持storm和flink吗？
[附加信息]
浏览 589提问于2018-09-10
1回答
我有两个来自两个Kafka主题的流表，我想加入这些流并对所连接的数据执行聚合功能。需要使用滑动窗口连接流。在连接和窗口数据时，我将得到一个错误Rowtime attributes must not be in the input rows of a regular join. As a workaround you can cast the time attributes of input tables to TIMESTAMP before.。
下面是代码片段
select cep.payload['id'] , ep.payload['id'] ,
浏览 25提问于2022-05-24得票数 0
1回答
如何将PyFlink性能与Flink + Scala进行比较？
大照片。我们的目标是建立冷和热层的Lambda架构。冷(批)层将用Apache (PySpark)实现。但是对于热(流)层有不同的选择:火花流或Flink。
因此，Apache是纯流的，而不是Spark的微批，我倾向于选择Apache。但我唯一关心的是PyFlink的性能。PySpark流的延迟会更少吗？它比Scala编写的Flink代码慢吗？在什么情况下更慢？
提前谢谢你！
浏览 3提问于2021-11-05得票数 1
回答已采纳
我在使用Flink连接两个流的上下文中查看，并希望了解这两个流之间的区别以及影响Flink处理它们的方式。
作为一个相关的问题，我还想了解CoProcessFunction与KeyedCoProcessFunction有何不同。
浏览 2提问于2021-02-17得票数 1
回答已采纳
我使用的是使用Java的Flink表API，其中我想将DataSet转换为DataStream .以下是我的代码：
TableEnvironment tableEnvironment=new TableEnvironment();
Table tab1=table.where("related_value < 2014").select("related_value,ref_id");
DataSet<MyClass>ds2=tableEnvironment.toDataSet(tab1, MyClass.class);
DataStream&
浏览 4提问于2016-05-13得票数 4
我在线程"main“中得到了一个异常:类org.apache.flink.state.api.runtime.SavepointLoader试图访问受保护的方法org.apache.flink.runtime.state.filesystem.AbstractFsCheckpointStorage.resolveCheckpointPointer(Ljava/lang/String;)Lorg/apache/flink/runtime/state/CompletedCheckpointStorageLocation；(org.apache.flink.state.api.runtime
浏览 0提问于2020-01-20得票数 1
回答已采纳
2回答
我们知道Flink是一个真正的实时流处理引擎，它可以在记录到达时处理记录，我们也知道火花流是一个微批量流处理引擎。
然而，我们也知道星星之火发布了结构化的流媒体，它怎么样？它是否也是一个真正的实时流处理引擎，就像Flink一样，它可以在记录到达时立即处理记录，而不是微批处理，或者仍然使用微批处理模式？
浏览 3提问于2017-06-27得票数 3
回答已采纳
我使用Kafka连接器创建了一个purchases表(使用Flink )：
CREATE TABLE purchases (
country STRING,
product STRING
) WITH (
'connector' = 'kafka',
'topic' = 'purchases',
'properties.bootstrap.servers' = 'kafka:29092',
'value.format' = 'json'
浏览 5提问于2021-12-13得票数 1
回答已采纳
我试着理解如何在Flink工程中CEP。例如，来自pages：和。
我想知道Flink CEP应用程序中(Scala)代码的基本结构是什么？
应该使用哪些(必需的)部件、类、函数/方法等，以及按什么顺序使用？
我认为基本部分可以是: 1.模式定义2.模式流定义(或者可以使用许多并行流吗?) 3.选择定义是正确的吗？
如何使用"select part“以及如何查找有关它的更多信息？它看起来相当复杂。
浏览 0提问于2018-03-06得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
