在Flink中，当我们有两个或更多操作符输出相同数据类型的记录时，我们可以重用数据输出数据类型的OutputTag吗？示例： OutputTag<A> sideOutputTag = new OutputTag<A>("side-output") {};
ProcessFunction1 processFunction1
浏览 69提问于2020-09-10得票数 3
回答已采纳
如何根据不同的用例展开相同的keyedStream并应用过滤器，而无需在过滤结束时创建新的keyedStream？示例： DataStream<Event> streamFiltered = RabbitMQConnector.eventStreamObject(env) .name("Event Mapper")
.assignTimestampsAndWatermarks(new PeriodicExtractor(
浏览 28提问于2020-09-04得票数 0
extractTimestamp(t: SensorReading, l: Long): Long = t.timestamp )
浏览 2提问于2021-03-16得票数 0
回答已采纳
2回答
> timedOutPatternStream = CEP.pattern(inputStream, simplePattern.within(Time.seconds(10))); SingleOutputStreamOperator<String> timedOutNotificationsStream
浏览 0提问于2018-05-15得票数 2
回答已采纳
我有一个工作，它使用Flink的端输出功能将数据写入不同的kafka接收器。在IDE中运行时，侧输出将获取写入它的数据，而在Flink集群上运行时，则不会。知道为什么吗？以下是代码的示例：
final OutputTag<SideOutputObject> sideOutputTag = new OutputTag<SideOutputObject>("side-output-tag
浏览 1提问于2017-11-29得票数 2
回答已采纳
关于flink的并行性，我有几个问题。这是我的设计：1.)传递-p 2只配置任务并行性，即任务(如map(deserialize))将被拆分成的最大并行实例数。2.)在我看来，这似乎是矛盾的/令人困惑的:即使将并行性设置为1，不同的任
浏览 4提问于2017-04-13得票数 3
回答已采纳
Apache的新手，我尝试过滤以字母"N“开头的单词，并得到输出，但是我如何获得不以单词"N”开头的单词，这是我正在使用的代码import org.apache.flink.api.common.functions.FlatMapFunction;import o
浏览 1提问于2020-05-12得票数 0
回答已采纳
我有一个Flink应用程序，它读取任意AVRO数据，将其映射到RowData，并使用几个FlinkSink实例将数据写入冰山表。为此，我使用了侧输出:当我将数据映射到RowData时，我使用ProcessFunction将每条消息写入特定的OutputTag。类似于：
final.process(new Pro
浏览 18提问于2022-10-18得票数 0
回答已采纳
我正在尝试使用BroadcastState设置一个项目，但由于某些原因，我在尝试运行它时遇到以下错误： org.apache.flink.streaming.runtime.tasks.StreamTaskException:无法序列化运算符对象类org.apache.flink.streaming.api.operators.co.CoBroadcastWithNonKeyedOperator.传入和输出的对象(SampleInput和Token)是非常简单的avro生成的pojos，有两个或三个字段，我试着把BroadcastProcessFunctio
浏览 58提问于2019-03-13得票数 0
回答已采纳
我是Flink的新手，我正在使用apache flink进行模式匹配，其中模式列表处于广播状态，并迭代通过processElements函数中的模式来找到匹配的模式，我正在从数据库中读取此模式，这是一个准时的活动BasicTypeInfo.STRING_TYPE_INFO,
public final static OutputTag<Tuple2<String, String>> unMatchedSideOutput
浏览 38提问于2020-06-08得票数 0
当前解决方案：stream.keyBy(type)| DOG | 20 | (only id 2 as been 'summed')
Id 4和5仍在flink弗林克有办法做到这一点吗？我会想象，我们将有一个聚合函数，它可以对金额进行求和，但也会输出每条
浏览 0提问于2018-08-23得票数 2
回答已采纳
3回答
我在Flink中做实时流，其中Kafka是消息队列。我正在申请120秒的EventTimeSlidingWindow。滑动1秒。我还在事件时间的每一秒插入水印。现在在我的情况下，Flink简单地丢弃了它各自的水印之后的消息。filnk是否提供了任何机制来处理这种延迟消息，比如维护单独的窗口？我也浏览了文档，但我没有弄清楚。
浏览 19提问于2017-01-19得票数 3
回答已采纳
我已经开发了一个Flink程序，可以读取Twitter上的推文，并将它们推送到Kafka上。然后，它从Kafka获取tweet并对其进行处理。附加的图像是从Flink Web UI中拾取的。程序是正确的吗？DataStream<Tuple3<String, String, Integer>> edgesStream = termsStream.getSideOutput(TweetParse.outputTag
浏览 1提问于2018-06-30得票数 1
我有一个小的Flink应用程序： public class App {
public static final OutputTag<String> numberOutputTag = new OutputTag我已经阅读了文档，但我看不到任何可以解释为什么会是这样的东西。我很好奇我是不是错过了什么或者做错了什么。gist.github.com/baelec/95f41d875dda0a2806a0fb9b9313b90e 如果您更喜欢下载示例项目，这里有一个存储库：https:/
浏览 15提问于2020-09-13得票数 1
1回答
让我以一种通用的方式开始，看看我是否错过了一些概念:我有一个流flink作业，我从该作业创建了一个保存点。(batchMode) {} flink.addKafkaSource(topicName)stream.process(new ProcessorWithKeyedState())
CassandraSink.addSink(stream) 只要我在没有保存点的情况下运行作业，这就可以很好地工作<i
浏览 51提问于2021-08-03得票数 0
回答已采纳
在使用Flink的Hive连接器查询Hive时，我有以下异常：Distinct without an aggregation。Hive版本：2.1.1select devid as pdevid,
count(distinct vtype) as vip_type_transat org.apache.flink.client.program.Packag
浏览 21提问于2022-07-08得票数 0
ProcessFunctionB); //stream was processed by function A 如果ProcessFunctionA更改了数据，ProcessFunctionB会看到更改吗？
浏览 120提问于2021-04-24得票数 1
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
腾讯会议
云直播
对象存储活动推荐
腾讯云开发者
