如何在Flink中连接2个以上的流？
在Flink中连接两个以上的流可以通过使用Flink的连接操作符和联合操作符来实现。
- 连接操作符（Connect）：连接操作符用于将两个流进行连接，连接后的结果是一个包含两个输入流的联合流。连接操作符适用于连接不同数据类型的流。
- 概念：连接操作符是将两个流进行连接，通过共享相同的键值，将两个流进行合并成一个联合流。
- 优势：连接操作符能够将不同类型的流连接在一起，并保留各自的数据结构。
- 应用场景：适用于需要将不同类型的数据流进行合并处理的场景，如合并实时日志数据和传感器数据等。
- 推荐的腾讯云相关产品：腾讯云Flink计算引擎
- 产品介绍链接地址：https://cloud.tencent.com/product/flink
- 联合操作符（Union）：联合操作符用于将两个或多个具有相同数据类型的流进行联合，联合后的结果是一个包含所有输入流的联合流。联合操作符适用于连接相同数据类型的流。
- 概念：联合操作符将两个或多个相同类型的流进行联合，将所有输入流的数据合并成一个联合流。
- 优势：联合操作符可以将多个相同类型的流合并为一个流，方便进行统一的处理。
- 应用场景：适用于需要将多个相同类型的数据流进行合并处理的场景，如合并多个传感器数据流进行实时分析等。
- 推荐的腾讯云相关产品：腾讯云Flink计算引擎
- 产品介绍链接地址：https://cloud.tencent.com/product/flink
通过使用连接操作符和联合操作符，可以在Flink中连接两个以上的流，实现流之间的数据处理和分析。以上为Flink中连接多个流的基本概念、优势、应用场景以及推荐的腾讯云相关产品和产品介绍链接地址。
相关·内容
1回答
我正在对flink中的两个表执行一个左外部联接，下面给出了代码，显示了一个异常，使用相同两个表的内部连接工作良好，能够转换为Datastream。
Table table = customerTable.leftOuterJoin(contactTable,$("cust_custcode")
.isEqual($("contact_custcode")))
.select($("customermessage"), $("contactmessage"));
例外情况是: org.apache.flink.table.api.T
浏览 8提问于2022-08-11得票数 0
Flink流支持内部连接表达式，如window- join，interval-join.但不支持左联接/完全联接表达式。当然，窗口-cogroup表达式可以实现相同的语义，即使事件立即加入，也必须等待完全窗口大小的时间。我的问题是：
如何从设计角度解释Flink流不支持左连接/完全连接表示？如何通过Flink DataStream API (如果能够立即转发连接事件)来实现它？是否有一种方法可以扩展Flink DataStream API以支持左连接：
.leftJoin()
.where()
.window()
.apply()
浏览 3提问于2021-08-15得票数 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
目前，我们使用Flink FsStateBackend检查点并设置fileStateSizeThreshold，将写入HDFS上avro/json文件的数据大小限制在128 on。在检查点操作延迟一定时间后也关闭文件。
由于我们在新项目中没有使用高级Flink特性，所以我们希望使用Kafka Connect HDFS Connector中的Kafka流直接将消息写入hdfs (而不是旋转Flink)。
但是，我无法找到限制卡夫卡连接器中hdfs文件的文件大小的选项，除了flush.size，它似乎限制了记录的#。
如果连接器上没有设置，人们如何以另一种方式管理hdfs上流数据中的文件大小？
浏览 0提问于2018-08-08得票数 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
回答已采纳
我有两个来自卡夫卡的事件源，一个来自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
回答已采纳
1回答
我的Cassandra接收器配置如下：
ClusterBuilder secureCassandraSinkClusterBuilder = new ClusterBuilder() {
@Override
protected Cluster buildCluster(Cluster.Builder builder) {
return builder.addContactPoints(props.getCassandraClusterUrlAll().split(","))
浏览 0提问于2018-04-04得票数 1
回答已采纳
我在使用Flink连接两个流的上下文中查看，并希望了解这两个流之间的区别以及影响Flink处理它们的方式。
作为一个相关的问题，我还想了解CoProcessFunction与KeyedCoProcessFunction有何不同。
浏览 2提问于2021-02-17得票数 1
回答已采纳
我有一个像这样的流：<_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
我认为斯派克比弗林克优越。下面是我的研究。
我看到星火的大部分功能都包含在Flink中，除了星火的“公平调度”。我试着搜索并查看Flink文档，但是没有运气。
另外，如果你看到吉顿，Apache的流行度几乎是Flink的两倍(明星和叉子的数量)。所以我很想知道为什么弗林克没有星火那么受欢迎。
我还看到写入Flink的连接器数量比Spark的连接器数量少/维护少。(例如: mongodb)这是否意味着Flink尚未成熟或获得市场吸引力？
以上问题的答案将帮助我们决定合适的技术。
编辑-1
在阅读了一些答案之后，我会给出更多的信息。
这将主要用于批处理。实时流可能占用例的1
浏览 4提问于2022-04-21得票数 -2
2回答
如何在Flink转换中更新配置？
、、、、
给定一个Flink流作业，该作业将map()操作应用于流。
此map()操作从某些属性读取其配置，并相应地映射数据。例如，配置指定读取属性"input"，并使用不同的属性名"output“将其写入流。这个已经很好了。
现在，配置发生了变化，例如，转换将为输出使用不同的属性名。
因此，我正在寻找一种方式，让所有的Flink任务重新阅读一个新的配置在运行时。
有没有可能
将KafkaSourcewait挂起，直到管道耗尽(flush)trigger集群中的所有任务，以重新读取配置文件(coordinated)resume (coordinated)resume KafkaS
浏览 13提问于2022-10-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
回答已采纳
我想用java编写一个通用flink作业，它可以接受任何SQL-SELECT查询，对SQL-数据库运行它，并将其写入Elasticsearch索引中。
我必须解决的问题之一是为JDBC连接创建一个DataSource。我想使用。我遵循了中的例子。
问题是，必须指定泛型类型DataSource类型。我只能使用Tuple类型，因为JDBCInputFormat泛型类型OUT扩展了Tuple。但我不知道编译时我将使用哪个Tuple。
我是不是解释错了？
我还可以使用另一个jdbc InputFormat吗？
是否有方法将Tuple指定为泛型类型？
我使用java 7和0.10.2
浏览 1提问于2016-02-24得票数 0
回答已采纳
我计划动态缩放一个Flink应用程序。该应用程序使用kafka-flink连接器消耗来自Kafka的事件。
因为应用程序的“热身”只需几分钟(缓存.)而改变并行级别涉及重新启动，我更喜欢提交(放大)或交替杀死(缩小)任务，而不是改变并行级别。
我想知道，从性能、逻辑和执行计划来看，这种方法和Flink内置的并行执行有什么不同吗？
换句话说，10个相同的Flink任务与一个并行级别= 10的任务( env.setParallelism( 10 ) )有什么区别？
浏览 1提问于2019-09-25得票数 0
回答已采纳
这是一个关于连接的键控流的非常基本的问题。
如果我有两个流，具有共享相同逻辑键的相关事件，并且这些流正在连接(使用键逻辑连接)，并且都以> 1的并行性运行，那么Flink如何保证来自不同流的两个具有相同逻辑键的事件最终在同一并行操作符实例中？
这是一个虚构的关于医院病人流的例子-体温流和心跳流。我们希望通过患者id使用ConnectedStream和CoFlatMapFunction将这两个流连接起来。
DataStream<PatientTemperature> temperatureStream = ..
DataStream<HeartbeatStream>
浏览 79提问于2018-06-03得票数 0
回答已采纳
下面是我使用flink sql API连接两个表的代码 tEnv.createTemporaryView("A", streamA,"speed_sum,cnt,window_start_time,window_end_time");
tEnv.createTemporaryView("B",streamB,"speed_sum,cnt,window_start_time,window_end_time");
String execSQL1 = "select A.speed_sum+COALESCE(B.speed_
浏览 211提问于2021-10-27得票数 1
我正在为流分析方案评估Flink，并且还没有找到关于如何实现我们今天在遗留系统中所做的一种ETL设置的足够信息。
一个非常常见的场景是，我们使用了关键的、慢吞吐量的元数据流，我们希望使用这些数据流在高吞吐量数据流上进行丰富，如下所示：
这就引出了两个关于Flink的问题:如何通过缓慢更新时间窗口重叠但不相等的流来丰富快速移动的流(元数据可以存活数日，而数据可以存活几分钟)？如何有效地将多个(最多10个)数据流与Flink连接起来，比如一个数据流和9个不同的富集流？
我知道我可以使用非窗口的外部ETL缓存来实现我的ETL场景，例如使用Redis (这是我们今天使用的)，但我想看看Fli
浏览 3提问于2017-11-21得票数 1
回答已采纳
1回答
在Flink中广播“加入”
、、、、
是否有任何方法，我可以使用广播连接在FLINK与我在火花使用的方式。我正在使用联接，但是数据很大，所以我需要广播连接。
谢谢
浏览 0提问于2019-10-17得票数 0
回答已采纳
3回答
我试图从oreilly书“用Apache进行流处理”和flink项目中运行第一个示例。每个错误都是不同的
例如，在书中，flink项目的NoClassDefFound错误示例给出了java.net.ConnectException:连接拒绝(连接拒绝)，但是确实创建了一个flink作业，参见屏幕截图。
详情如下
图书示例
java.lang.NoClassDefFoundError:scala/runtime/java8/JFunction1$mcVI$sp at io.github.streamingwithflink.chapter1.AverageSensorReadings$$ano
浏览 7提问于2020-04-05得票数 1
回答已采纳
我正在开发一个应用程序，它将一些文件上传到s3桶，稍后，从s3桶中读取文件，并将其推送到数据库。
我使用Flink 1.4.2和fs.s3a API从s3桶读取和写入文件。
上传文件到s3桶没有任何问题，但是当我的应用程序的第二阶段--从s3读取这些上传的文件--启动时，我的应用程序会抛出错误
Caused by: java.io.InterruptedIOException: Reopen at position 0 on s3a://myfilepath/a/b/d/4: org.apache.flink.fs.s3hadoop.shaded.com.amazonaws.SdkClient
浏览 0提问于2019-06-21得票数 1
我使用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内存管理。在少数博客中，我找到了“内存管理器池”和"Rocksdb“。我正在使用rocksdb，并且假设我的所有状态都存储在该db中。
这是我的怀疑..。
如何在流中处理内存管理过程？
流和批处理中的内存管理有什么区别？
“内存管理器池”和“后端状态(Rcokdb)”的区别
在流中，您所说的"Flink托管内存“是什么意思？是否包括RacksDb缓存和缓冲区所需的内存？
浏览 9提问于2020-05-01得票数 1
1回答
当我的制作人把所有的消息都流到Kafka之后，在Flink完成处理之后，我希望能够终止Flink作业，这样它就不会继续运行了，这样我就可以知道Flink什么时候完成了所有的数据处理。我也不能使用批处理，因为我需要Flink运行平行于我的Kafka流。
通常，Flink在一个isEndOfStream类中使用DeserializationSchema方法来查看它是否应该提前结束(返回方法中的true将自动结束作业)。但是，当使用Flink作为来源时，新的KafkaSource类不赞成在反序列化器中使用isEndOfStream方法，不再检查它是否应该结束流。还有其他方法可以提前终止Flink工作
浏览 13提问于2022-10-03得票数 0
我需要在flink中使用foreach。有没有人可以举例说明如何在flink中使用map()来实现spark中的foreach？
浏览 1提问于2016-12-15得票数 0
我正在尝试使用Apache连接twitter，使用官方文档进行流传输。
虽然我能够成功地使用api解析示例TwitterExampleData，但是当我试图使用相同的方法连接真实的数据时，我就得到了ClassNotFoundException。
java.lang.NoClassDefFoundError: org/apache/flink/twitter/shaded/com/google/common/base/Preconditions
at com.twitter.hbc.core.endpoint.BaseEndpoint.<init>(BaseEndpoint.java
浏览 6提问于2017-10-03得票数 2
回答已采纳
我试图用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
我有一个带有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
鞋(一个事件)被定义为它的颜色和isLeft (如果鞋是左腿，那么isLeft=true，否则false)。
Tuple2<String, Boolean> leftBlueShoe = Tuple2.of("blue", true);
Tuple2<String, Boolean> rightBlueShoe = Tuple2.of("blue", false);
// unbounded stream of shoes is as follows
DataStream<Tuple2<String, Boolean>
浏览 3提问于2020-06-18得票数 0
回答已采纳
1回答
我使用的是Flink 1.12。我想读取csv，并根据处理时间进行翻滚窗口分组。 代码如下，但是查询sql_tubmle_window没有输出，我想知道问题出在哪里 import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.types.Row
object Sql017_ProcessTimeAttributeDDLTest {
def main(args: Array[String]): Unit = {
浏览 5提问于2021-05-02得票数 0
回答已采纳
来自Flink的官方文档：
示例代码如下：
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
...
val orangeStream: DataStream[Integer] = ...
val greenStream: DataStream[Integer] = ...
orangeStream
.keyBy(elem => /* select key */)
浏览 11提问于2020-02-17得票数 0
回答已采纳
为了在现有源和汇的基础上构建源和汇(而不是从头开始或使用更多样板)，我想：
将转换函数f:A->B附加到Source<A>以获得Source<B> (对应于FP圈中的映射)
准备将函数f:B->A转换为Sink<A>以得到Sink<B> (对应于FP圈中的逆图)
在Scala中，JSON库(如Circe)通常使用这种模式：
这个模式适用于手头的情况(Flink源和接收器)吗？
否则，解决这个问题的建议是什么？
作为该问题的一个特殊实例，您可以考虑以下示例：
您有一个现有的DataStream连接器(源
浏览 3提问于2022-07-19得票数 1
支持多个运行程序后端，包括Apache和Flink。我很熟悉Spark/Flink，我正试图看到Beam批量处理的利弊。
从的角度来看，它觉得它非常类似于本机Spark/Flink的等价物，可能有一个稍微冗长一些的语法。
我目前不认为选择梁而不是火花/Flink来完成这样的任务有多大的好处。到目前为止，我能提出的唯一意见是：
支持:对不同执行后端的抽象。
Con:这种抽象的代价是减少了对在Spark/Flink中执行什么的控制。
是否有更好的例子突出梁模型的其他利弊？是否有任何关于失控如何影响性能的信息？
请注意，我并不是在请求流方面的差异，流方面的部分内容在中进行了介绍，并在
浏览 4提问于2017-04-24得票数 108
回答已采纳
我有一个特定的任务，使用Apache Flink和一些额外的逻辑将两个数据流连接到一个聚合中。
基本上，我有两个数据流:事件流和所谓的meta-events流。我使用Apache Kafka作为消息主干。我试图实现的是根据aggregation/window中给出的信息触发meta-event到评估。基本情况是：
事件数据流( Data of events )开始根据某些键在某个聚合或窗口中不断积累Type A;The记录的记录；(元事件数据流)使用给定的键接收新的meta-event，这也定义了将在数据流事件流中发出的事件总数。，构成步骤3的事件数成为聚合的触发条件。当具有给定键的Type
浏览 6提问于2020-07-06得票数 0
回答已采纳
在我看来，Processing Time Temporal Join用于流和外部数据库，并且总是基于联接条件在外部数据库中的最新值join。此外，当Processing Time Temporal Join时使用the external table is not feasible to materialize the table as a dynamic table within Flink.
类似地，Lookup Join用于流和外部数据库，并且始终基于联接条件在外部数据库中look up值。
Lookup Join会在Flink中实现外部数据库表吗？他们之间有什么区别？
浏览 2提问于2022-02-21得票数 3
回答已采纳
我试图运行一个flink蒸汽程序，使用卡夫卡连接器(最新的通用连接器)。作业在IntelliJ上运行时没有任何问题，但是当我使用sbt包将代码提交到jar中时，会给我以下错误。java.lang.ClassNotFoundException: org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase
我还使用了使用传统IntellIJ选项构建的jar，但仍然得到了上述错误。
浏览 0提问于2020-02-28得票数 0
我们感兴趣的是从连接到常规的Flink流应用程序，最好使用Table。这样做的想法是查阅Statefun在Flink中注册的表，是否有此可能，如何做才是正确的方法？
到目前为止，我的想法是在一些主函数中初始化我的表流，并注册一个有状态函数提供程序来连接到该表：
@AutoService(StatefulFunctionModule.class)
public class Module implements StatefulFunctionModule {
@Override
public void configure(Map<String, String> globalC
浏览 2提问于2020-04-11得票数 2
我试图用Flink的KafkaSource运行一个简单的测试程序。我使用的方法如下：
Flink 0.9
Scala 2.10.4
Kafka 0.8.2.1
我遵循docs测试KafkaSource (添加了依赖项，捆绑了和中的卡夫卡连接器flink-连接器- Kafka )。
下面是我的简单测试程序：
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka
object TestKafka {
def main(args: Ar
浏览 3提问于2015-07-15得票数 5
我试图连接卡夫卡与Flink和运行通过sql-client.sh。但是，无论我如何处理.yaml和库，我都会得到错误：
Exception in thread "main" org.apache.flink.table.client.SqlClientException: Unexpected exception. This is a bug. Please consider filing an issue.
at org.apache.flink.table.client.SqlClient.startClient(SqlClient.java:201)
浏览 46提问于2021-11-30得票数 0
回答已采纳
1回答
如果我连接Flink中的键流和非键流会发生什么？例如；
val stream1 = source.keyBy(_.customerId)
val stream2 = source2.map(operation()...)
stream1.connect(stream2).process(processData).addSink(kafkaSink)
这里的情况是，我试图将一组规则(无键流，stream2)应用到另一个流(stream1)，我通常在广播规则流，但是广播的大小越来越大。所以，我只想知道，如果我不广播其中一个流，并将它们按并行性、状态等连接起来，会发生什么呢？
浏览 8提问于2022-08-07得票数 0
我有apache访问日志文件，我想在cassandra表中存储每个页面的访问计数(总数/每天/每小时)。
我正在尝试使用kafka连接从日志文件流到kafka主题。为了在Cassandra中增加指标计数器，我可以再次使用Kafka Connect吗？否则，这里应该使用哪些其他工具，例如kafka streams，spark，flink，kafka connect等？
浏览 0提问于2017-10-11得票数 0
1回答
我是flink的新手，我们的用例处理无状态计算。读取事件、处理事件并保存到数据库中。但是Flink文档从来没有提到过无状态处理。任何示例存储库，用于查找无状态示例或文档。
最后，对于这个用例，哪个Flink模型有效？流应用程序或事件驱动应用程序。
浏览 21提问于2020-12-09得票数 2
我发现在jdbc连接器的部分只有DDL和yaml格式的配置，我不知道如何使用them.so。我问的是在写flink app jar.if时如何从jdbc数据源读取流数据。如果数据源中的数据发生了变化，流是否会更新。
浏览 86提问于2020-06-23得票数 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
1回答
我从flink开始，看一看。
据我理解，这个练习的目标是加入时间属性上的两个流。
任务：
这个练习的结果是Tuple2记录的数据流，每个不同的rideId都有一个数据流。您应该忽略终点事件，只在每次乘车开始时加入活动，并提供相应的车费数据。
结果流应打印为标准输出。
问题：，EnrichmentFunction如何能够加入这两个流？它怎么知道参加哪一场比赛呢？我希望它能缓冲多个交易会/骑行，直到一个即将到来的交易会/骑行有一个匹配的合作伙伴。
据我理解，它只会节省它所看到的每一次骑行/公平，并将它与第二次最好的骑行/交易会结合起来。为什么这是一个适当的加入？
提供的解决方案：
/*
浏览 0提问于2019-01-20得票数 4
回答已采纳
为什么Flink-HBase connector中没有HBaseReadStreamExample 似乎有一个读、写和WriteStream示例，但没有ReadStream。 从HBase到Flink的流读取是一种糟糕的做法吗？
浏览 21提问于2020-07-03得票数 1
回答已采纳
假设我有两个不同类型的DataStream：
val stream1: DataStream[(Int, Int, Int)] = ...
val stream2: DataStream[(Int, Int, Int, Int, Float)] = ...
如何将两个流写入单个文件
我试过不同的东西，但似乎没用。例如，我不能直接写成
stream1.writeAsText("path/to/file.txt").setParallelism(1)
stream2.writeAsText("path/to/file.txt").setParallelism(1)
浏览 0提问于2016-08-30得票数 0
回答已采纳
我是Apache Flink的新手。我必须过滤我计算机中的所有文件，并将它们写入一个文件。但在Flink中，似乎不可能写入现有文件。我该怎么处理呢？
浏览 33提问于2019-06-10得票数 0
1回答
我有一个DataStream<Tuple2<String, Double>> one和一个DataStream<Tuple2<String, Double>> second，其中第一个包含来自另一个的更多元素，并且它们具有不同的键。此外，数据流"two“基本上有一个键值对。因此，我希望将这些流连接起来，以便将第一个数据流的值与第二个数据流的常量值相除。如何在Apache Flink中做到这一点？这是通过连接的数据流完成的，还是另一种方式？
浏览 3提问于2020-02-28得票数 0
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
实时音视频
对象存储
云直播活动推荐
腾讯云开发者
