Flink流未完成
是指Apache Flink流处理框架中的一个特性，用于处理未完成的数据流。下面是对该特性的完善且全面的答案：
概念： Flink流未完成是指在流处理过程中，由于各种原因导致的数据流未能完整地被处理的情况。这些未完成的数据流可能是由于网络延迟、故障、数据丢失等原因引起的。
分类： Flink流未完成可以分为两类：未完成的事件时间数据和未完成的处理时间数据。未完成的事件时间数据是指在事件时间窗口中，由于数据延迟或乱序到达而导致的数据流未能完整地被处理。未完成的处理时间数据是指在处理时间窗口中，由于数据延迟或其他原因导致的数据流未能完整地被处理。
优势： Flink流未完成的特性使得开发人员能够更好地处理延迟和乱序数据，提高数据处理的准确性和完整性。通过对未完成的数据流进行处理，可以避免数据丢失和错误的计算结果，提高数据处理的可靠性和稳定性。
应用场景： Flink流未完成的特性在以下场景中非常有用：
- 实时数据分析：对于需要实时分析大规模数据的场景，Flink流未完成可以确保数据的完整性和准确性，提供准确的实时分析结果。
- 金融交易处理：在金融领域，对于交易数据的处理需要保证数据的完整性和顺序性。Flink流未完成可以处理延迟和乱序的交易数据，确保交易数据的准确性和一致性。
- 物联网数据处理：物联网设备产生的数据通常具有高度的实时性和时序性。Flink流未完成可以处理物联网设备发送的延迟数据，确保数据的完整性和及时性。
推荐的腾讯云相关产品： 腾讯云提供了一系列与流处理相关的产品和服务，以下是其中几个推荐的产品：
- 腾讯云流计算Oceanus：腾讯云的流计算服务，提供高性能、低延迟的流处理能力，支持实时数据分析和处理。
- 腾讯云消息队列CMQ：腾讯云的消息队列服务，可以用于实时数据的传输和处理，支持高并发、可靠性强的消息传递。
- 腾讯云云数据库TDSQL：腾讯云的云数据库服务，提供高可用、可扩展的数据库解决方案，适用于存储和管理实时数据。
产品介绍链接地址：
- 腾讯云流计算Oceanus：https://cloud.tencent.com/product/oceanus
- 腾讯云消息队列CMQ：https://cloud.tencent.com/product/cmq
- 腾讯云云数据库TDSQL：https://cloud.tencent.com/product/tdsql
相关·内容
我只是想了解在Flink中使用CoProcessFunction的用例。举例说明会帮助我更好地理解这个概念。
浏览 21提问于2019-07-28得票数 1
回答已采纳
这是我的问题:在发送停止命令时，filnk是否会完成接收过程，并将.inprogress文件重命名为part-x-x文件？
我发现我的flink任务(使用flink-1.9.1)不会将.inprogress文件重命名为part-x-x文件。但我读了源代码，上面写着
org.apache.flink.client.program.ClusterClient#stopWithSavepoint
* Stops a program on Flink cluster whose job-manager is configured in this client's configuration.
浏览 5提问于2020-05-31得票数 0
我试图用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
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
回答已采纳
我有一个像这样的流：<_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回答
我正在尝试将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
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 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
回答已采纳
这段代码将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
回答已采纳
我试图在flink服务器上运行一些代码，但无法
下面是我的代码
DataStream<UserInfo> keyedStream = executionEnvironment
.addSource(new UserDataSource());
keyedStream.assignTimestampsAndWatermarks(new MessageWaterEmitter());
tableEnv.registerDataStream("test", keyedStream, "us
浏览 0提问于2020-12-30得票数 0
回答已采纳
我有一个带有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
我在使用Flink连接两个流的上下文中查看，并希望了解这两个流之间的区别以及影响Flink处理它们的方式。
作为一个相关的问题，我还想了解CoProcessFunction与KeyedCoProcessFunction有何不同。
浏览 2提问于2021-02-17得票数 1
回答已采纳
我使用intellij、maven 3和flink 1.15.1编写有状态流作业。我正在尝试为我的自定义KeyedProcessFunction编写单元测试，并尝试遵循文档以及添加提到的的依赖项。我对使用KeyedOneInputStreamOperatorTestHarness类很感兴趣，但是在我的任何依赖项中都找不到这个类(下面发布的修改后的pom )。我唯一能找到这门课的方法就是
<dependency>
<groupId>org.apache.flink</groupId>
<ar
浏览 8提问于2022-08-01得票数 1
回答已采纳
为什么Flink-HBase connector中没有HBaseReadStreamExample 似乎有一个读、写和WriteStream示例，但没有ReadStream。 从HBase到Flink的流读取是一种糟糕的做法吗？
浏览 21提问于2020-07-03得票数 1
回答已采纳
我有两个流。第一个是基于时间的流，我使用countTimeWindow接收前10个数据点来计算STAT值。我手动使用了变量cnt来只保留第一个窗口，并过滤了剩余的值，如下面的代码所示。 然后，我想使用这个值来过滤主流，以便获得比我在窗口流中计算的stat值更大的值。 然而，我不知道如何合并或计算这两个流来实现我的目标。 我的场景是，如果我将第一个stat值转换为广播变量，然后将其传递给主流，以便能够根据广播变量中的stat值过滤传入的值。 下面是我的代码。 import com.sun.org.apache.xpath.internal.operations.Bool;
import org
浏览 10提问于2019-02-12得票数 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
回答已采纳
我在CheckpointedFuntion中使用运算符状态，但是在初始化MapState时遇到了NullPointerException：
public void initializeState(FunctionInitializationContext context) throws Exception {
MapStateDescriptor<Long, Long> descriptor
= new MapStateDescriptor<>(
"state",
TypeInfor
浏览 0提问于2018-03-18得票数 0
Spark有mapPartition API，而Flink DataStream API没有。有谁能帮助解释原因吗？我想要做的是在Flink上实现一个类似于星火reduceByKey的API。
浏览 2提问于2015-10-28得票数 7
回答已采纳
我正在为流分析方案评估Flink，并且还没有找到关于如何实现我们今天在遗留系统中所做的一种ETL设置的足够信息。
一个非常常见的场景是，我们使用了关键的、慢吞吐量的元数据流，我们希望使用这些数据流在高吞吐量数据流上进行丰富，如下所示：
这就引出了两个关于Flink的问题:如何通过缓慢更新时间窗口重叠但不相等的流来丰富快速移动的流(元数据可以存活数日，而数据可以存活几分钟)？如何有效地将多个(最多10个)数据流与Flink连接起来，比如一个数据流和9个不同的富集流？
我知道我可以使用非窗口的外部ETL缓存来实现我的ETL场景，例如使用Redis (这是我们今天使用的)，但我想看看Fli
浏览 3提问于2017-11-21得票数 1
回答已采纳
我有两个来自卡夫卡的事件源，一个来自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
回答已采纳
我试图在AWS EMR上部署我的flink作业(版本5.15和Flink 1.4.2)。然而，我无法从我的流中得到任何输出。我试着创造一个简单的工作：
object StreamingJob1 {
def main(args: Array[String]) {
val path = args(0)
val file_input_format = new TextInputFormat(
new org.apache.flink.core.fs.Path(path))
file_input_format.setFil
浏览 2提问于2018-07-19得票数 0
回答已采纳
我正在开发一个POC，将现有的Flink应用程序/拓扑转换为使用KafkaStreams。我的问题是关于部署。
具体来说，在Flink中，可以向flink安装添加"Worker节点“，然后为拓扑添加更多的并行化，以跟上不断增长的数据速率。
如何随着数据速率的增加而增加KStreams容量？KStreams会自动处理这个问题吗？我是否启动更多的进程(ala微服务)？
还是我错过了这里的大局？
浏览 4提问于2016-08-17得票数 3
回答已采纳
1回答
我有一个定制的flink Source，还有一个SerializableTimestampAssigner，它将事件时间戳分配给源发出的记录。由于底层数据存储的性质，源可能会发出无序的记录，但是在批处理模式下，我希望在任何操作符处理这些记录之前，Flink会根据事件时间戳对这些记录进行排序。
摘自Flink 的执行模式：
在批处理模式中，在预先知道输入数据集的情况下，不需要这样的启发式，因为至少可以按时间戳对元素进行排序，以便按时间顺序处理元素。
然而，情况似乎并非如此。如果我用时间戳分配器在Source (StreamExecutionEnvironment.fromSource)中
浏览 1提问于2022-04-22得票数 1
回答已采纳
我使用的是运行在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
我正在尝试写案例类序列化到AVRO到HDFS使用RollingSink在Flink。为了让avro文件可以被HDFS反序列化，我使用了包装FSDataOutputStream的DataFileWriter。当我试图在DataFileWriter和FSDataOutputStream之间同步以便关闭HDFS上的数据文件时，抛出异常，实际上我每隔一个文件就会得到数据。有没有办法在Flink writer实现中同步fs流和Avro编写器？
我尝试过使用DataFileWriter close()、flush()、sync()、fsync()，但都失败了。Sync方法似乎执行得最好。我也尝试过在wri
浏览 1提问于2016-03-24得票数 0
我想使用Apache来处理应用程序中的事件。我在独立JVM上的测试运行得相当好，尽管flink是一个非常大的依赖项。
我还试图让它在OSGi中运行，但现在放弃了，因为有许多依赖项。
所以我的问题是:我能做多小的Flink。目前，我尝试使用maven依赖于flink流-java。
不幸的是，这取决于或嵌入(只列出有问题的)：
flink-阴影-hadoop2 2
克里奥
动物园管理员
奈蒂
防波堤
apache http客户端
apache http核心
斯卡拉
阿克卡
杰克逊
它还看起来像几个jars一次又一次地嵌入相同的库。就像一些谷歌libs
浏览 4提问于2015-12-18得票数 1
我计划动态缩放一个Flink应用程序。该应用程序使用kafka-flink连接器消耗来自Kafka的事件。
因为应用程序的“热身”只需几分钟(缓存.)而改变并行级别涉及重新启动，我更喜欢提交(放大)或交替杀死(缩小)任务，而不是改变并行级别。
我想知道，从性能、逻辑和执行计划来看，这种方法和Flink内置的并行执行有什么不同吗？
换句话说，10个相同的Flink任务与一个并行级别= 10的任务( env.setParallelism( 10 ) )有什么区别？
浏览 1提问于2019-09-25得票数 0
回答已采纳
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
2回答
如何在Flink转换中更新配置？
、、、、
给定一个Flink流作业，该作业将map()操作应用于流。
此map()操作从某些属性读取其配置，并相应地映射数据。例如，配置指定读取属性"input"，并使用不同的属性名"output“将其写入流。这个已经很好了。
现在，配置发生了变化，例如，转换将为输出使用不同的属性名。
因此，我正在寻找一种方式，让所有的Flink任务重新阅读一个新的配置在运行时。
有没有可能
将KafkaSourcewait挂起，直到管道耗尽(flush)trigger集群中的所有任务，以重新读取配置文件(coordinated)resume (coordinated)resume KafkaS
浏览 13提问于2022-10-11得票数 0
正在尝试在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
无法使用StreamingFileSink并以压缩方式存储传入事件。 我正在尝试使用StreamingFileSink向S3写入无界事件流。在这个过程中，我想压缩数据，以便更好地利用可用的存储大小。 我写了一个压缩字符串写入器，通过从flink借用SequenceFileWriterFactory的一些代码。它会失败，并出现我在下面描述的异常。 如果我尝试使用BucketingSink，它会工作得很好。我使用BucketingSink编写压缩字符串，如下所示。再一次，我从其他的拉取请求中借用了这段代码。 import org.apache.flink.streaming.connectors
浏览 58提问于2019-05-16得票数 2
我使用的是使用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
我需要在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
2回答
我在流的世界里是个新手，在我的第一次尝试中我遇到了一些问题。 更具体地说，我正在尝试使用Flink在滑动窗口中实现计数和groupBy功能。 我在一个普通的DateStream中做过，但是我不能让它在WindowedStream中工作。 你对我该怎么做有什么建议吗？ val parsedStream: DataStream[(String, Response)] = stream
.mapWith(_.decodeOption[Response])
.filter(_.isDefined)
.map { record =>
(
浏览 18提问于2019-05-02得票数 1
2回答
我们知道Flink是一个真正的实时流处理引擎，它可以在记录到达时处理记录，我们也知道火花流是一个微批量流处理引擎。
然而，我们也知道星星之火发布了结构化的流媒体，它怎么样？它是否也是一个真正的实时流处理引擎，就像Flink一样，它可以在记录到达时立即处理记录，而不是微批处理，或者仍然使用微批处理模式？
浏览 3提问于2017-06-27得票数 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
回答已采纳
我正在运行flink 1.6.1的单节点flink集群上运行字数统计示例
并不断得到这个错误。
我使用的是beam版本2.8.0。
错误
jobmanager_1 | 2018-10-28 07:20:41,401 ERROR org.apache.flink.runtime.webmonitor.handlers.JarRunHandler - Exception occurred in REST handler.
jobmanager_1 | org.apache.flink.runtime.rest.handler.RestHandlerException: org.apa
浏览 0提问于2018-10-28得票数 1
我在线程"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
回答已采纳
1回答
我正在创建2个flink sql表，1个用于CSV文件系统，另一个用于kafka。其目标是持续监视文件系统文件夹，并将新的csv文件记录推送到kafka主题。但是我在下面写的查询，推送csv文件记录一次，flink作业进入“完成”模式，任何新的文件都不会被处理。请有人告诉我如何创建flink sql连续流与源和csv文件系统和目标为Kafka。
创建源表的Flink SQL
CREATE TABLE son_hsb_source_filesystem_csv_bulk(
file_name STRING,
start_time STRING,
oss_cell_id B
浏览 1提问于2021-03-24得票数 0
回答已采纳
1回答
查看Flink的文档和书籍，我对时间戳有疑问:如果流设置为事件-时间模式，这意味着时间戳具有进入Flink之前的源时间(甚至在通过消息队列(可能是Kafka)之前)，为什么Flink将时间戳作为元数据附加到记录？幻灯片3根据时间戳的用途列出了不同类型的时间戳：
如果时间戳位于事件内部，为什么要将该值传递给记录的元数据？另外，事件和记录之间的确切区别是什么？
浏览 5提问于2021-02-22得票数 0
1回答
我对flink很陌生，并试图将其部署到EMR集群上。我使用了3个节点集群(1个主服务器和2个从节点)进行默认配置。我没有做任何配置更改，并坚持默认配置。
我很想了解以下几点：
主从是如何相互通信的，因为我还没有在主节点中提到任何IP？
我可以在主节点(Path: /usr/lib/ flink )中看到flink库，但在从节点中找不到flink库。我的代码如何在从节点上执行？
如果需要，我将根据conf/flink-config.yml中的要求更改一些配置。除了这个之外，我还需要在主节点或从节点上做任何其他更改吗？
浏览 1提问于2018-05-10得票数 3
我正在尝试从Flink 1.12.x DataSet api迁移到Flink 1.14.x DataStream api。mapPartition在Flink DataStream中不可用。
我们的代码使用Flink 1.12.x DataSet
dataset
.<few operations>
.mapPartition(new SomeMapParitionFn())
.<few more operations>
public static class SomeMapPartitionFn extends RichMapPartition
浏览 9提问于2022-02-09得票数 0
我想从一个url中获得JSON数组形式的输入流。如何设置源，以便在apache flink中使用datastream.In short连续获得输入我希望从url连续获得json数据，而不关闭flink作业。
浏览 3提问于2018-03-08得票数 1
1回答
我使用的是Flink 1.12，我对何时可以执行表和数据集/数据集转换感到非常困惑。在下面的代码中，我希望将表内容打印到控制台，我尝试了以下3种方法，它们都会引发异常
table.toDataSetRow.print()table.toAppendStreamRow.print()table.print()
我会问如何将表内容打印到控制台，例如，使用print方法
import org.apache.flink.api.scala._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.a
浏览 2提问于2021-05-02得票数 0
回答已采纳
我是Apache的新手，我创建了一个简单的，我使用以下命令运行它：
python scripts/main.py \
--runner FlinkRunner \
--flink_master localhost:8081 \
--setup_file scripts/setup.py \
--environment_type EXTERNAL \
--environment_config localhost:50000 \
--database postgres \
--input /datas
浏览 8提问于2020-08-20得票数 4
假设我的Flink工作收到了一系列股票价格(举个例子)，如果某只股票跌破了一定的价格，我们就会发出警告。用户可以添加或删除这些警报条件。例如，用户abc@somemail.com创建了一条规则，以便在GME价格低于100美元时发出警告。我的Flink作业如何以可伸缩的方式动态跟踪所有这些警报条件？
我可以创建一个API，我的Flink作业可以调用这个API来获取所有更新的警报标准，但这意味着要多次调用API以保持每件事情都是最新的。
或者我可以用Flink表API创建一个永久表，一旦用户创建了一个新的警报标准，Flink作业就会更新该表。
这个用例的最佳实践是什么？
备注：
应以最小的延迟
浏览 4提问于2021-02-02得票数 0
回答已采纳
1回答
我有一个flink Row和列名，这个Row可以通过字段名或索引访问。我想用香草flink卡夫卡制片在JSON中融入卡夫卡。我该怎么做呢？目的地Json是否要求它将其沉到kafka？
浏览 7提问于2021-12-15得票数 1
我有一个非常简化的用例:我想使用Apache (1.11)从一个Kafka主题读取数据(让我们称之为source_topic)，计算其中的一个属性(称为b)，并将结果写入另一个Kafka主题(result_topic)。
到目前为止，我有以下代码：
from pyflink.datastream import StreamExecutionEnvironment, TimeCharacteristic
from pyflink.table import StreamTableEnvironment, EnvironmentSettings
def log_processing():
浏览 2提问于2020-11-16得票数 3
回答已采纳
我想设置Flink，这样它就可以将数据流从Apache转换并重定向到MongoDB。出于测试目的，我构建在flink流-Connectors.kafka示例()之上。
卡夫卡流是正确红色的链接，我可以映射他们等，但问题发生时，我想保存每一个接收和转换的信息到MongoDB。我发现的关于MongoDB集成的唯一例子是来自github的flink-mongodb测试。不幸的是，它使用的是静态数据源(数据库)，而不是数据流。
我认为应该有一些DataStream.addSink实现的MongoDB，但显然没有。
实现这一目标的最佳途径是什么？我是否需要编写自定义接收器函数，还是可能遗漏了什么？也许应
浏览 3提问于2016-02-02得票数 11
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
云直播活动推荐
腾讯云开发者
