我想加入一个大表，不可能包含在TM内存和一个流(kakfa)中。我在测试中成功地将table-api和datastream api结合在一起。我做了以下工作：stream .trigger
浏览 22提问于2021-03-03得票数 2
我实现了一个ProcessFunction，它使用Guava缓存来过滤传入的事件流。: The implementation of the ProcessFunction is not serializable.(StreamExecutionEnvironment.java:1560)at org.apache.flin
浏览 132提问于2019-04-02得票数 0
我正在通过。这里我们有一个选项来启用csv.ignore-parse-errors。有没有办法，我们可以把错误的消息，在DLQ的主题。
浏览 1提问于2022-03-01得票数 0
1回答
目标是处理带有table_name的元组和要插入的行列表。groupped.process(new ProcessFunction<Tuple2<String, List<JsonObject>>, Object>() {void processElement(Tuple2<String, List<JsonObject>> input,
ProcessFunction</e
浏览 2提问于2022-01-24得票数 0
我是Apache的新手，我尝试过滤以字母"N“开头的单词，并得到输出，但是我如何获得不以单词"N”开头的单词，这是我正在使用的代码import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple
浏览 1提问于2020-05-12得票数 0
回答已采纳
1回答
Flink表与蜂巢目录存储
、、、、
我有一个卡夫卡主题和一个蜂巢Metastore。我想加入卡夫卡话题中记录亚稳态的事件。我看到了Flink使用目录查询Hive的可能性。所以我看到了两种处理这个问题的方法：
使用表-Api，我将从kafka主题创建一个表，并将它与Hive目录连接起来。在第二种情况下，表是如何处理的？
浏览 5提问于2021-02-22得票数 1
我有两个(流)表，一个有事件时间列，另一个没有。我想使用Table加入这些程序，但是还没有想出一种同时保存时间戳信息的方法。考虑可以在Scala中执行的以下MWE：import org.apache.flink.streaming.api.functions.ProcessFunction(虽然奇怪的是，这只在从datastream创建表时显式提供.rowtime标记时才起
浏览 0提问于2018-08-15得票数 1
回答已采纳
/Create input data streams from kinesis data streams DataStreamString>) value -> value.uuid);
浏览 1提问于2019-08-02得票数 0
回答已采纳
我有一个管道，它使用Flink Global Window和基于事件时间的自定义触发器(从到达元素的时间戳开始)和Evictor，它从窗口中剪切不必要的元素并将其传递给ProcessFunction。类似于： public SingleOutputStreamOperator<Results> processElements(DataStream<Elements> inputStream) {}
public void executePipelineFlow(StreamExecuti
浏览 34提问于2021-11-18得票数 0
假设我有一个处理DataStream<X>并将返回发送到DB的函数，但是我需要从另一个源读取，并且在处理这个新的DataStream时，我需要在将DataStream<X>存储到DB中之前找到我可以生成的状态，并找到它进入DataStream<Y>的一个Id，然后触发一个操作。我的问题是：
例如，是否可以在Flink中使用Co-ProcessFunction来处理DataStream<X>中的转换结果，并在其中创建状态，同时处理Da
浏览 4提问于2020-11-23得票数 1
我有一个用例，其中我正在接收包含不同信息集的事件流，并希望对它们执行聚合。对于这些聚合中的每一个，都需要多个滚动窗口，例如: Daily、Weekly、Monthly、Yearly等。Daily B: 1 (Only for the day it was received) A: 7Monthly B: 4
浏览 3提问于2020-10-01得票数 1
介绍了如何使用Flink对无序流进行排序，但我更愿意使用DataStream API。将使用一个使用PriorityQueue来缓冲事件的ProcessFunction来实现这一点，直到水印表明它们不再是无序的，但是在RocksDB状态后端(问题是每次对PriorityQueue的访问都需要整个
浏览 2提问于2019-12-24得票数 5
我是Apache Flink的新手，我想了解DataStream和Table API之间的用例。请您帮我了解一下，什么时候选择Table接口而不是DataStream接口。据我所知，用Table API可以做的事情，也可以用DataStream API来做。这两个API有什么不同？
浏览 14提问于2018-08-01得票数 2
回答已采纳
我有一个表示为Set<Long>的自定义状态计算，当我的Datastream<Set<Long>>看到来自Kafka的新事件时，它将不断更新。我对所有的窗口和触发器操作有点迷惑，我一直收到以下错误。我只想知道如何将我的聚合流Datastream<Set<Long>>打印到标准输出或将其写回另一个kafka主题？
下面是抛出错误的代码片段。StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(env, bs
浏览 5提问于2020-03-04得票数 0
我得到了一个使用dataset api的项目，比如dataset.output(outputFormat)。outputFormat是userDefined(向neo4j写入批处理数据)，因此我希望保留它，但在flink use OutputFormat中找不到最新版本的表/sql。谢谢你的帮助。
浏览 38提问于2022-09-08得票数 0
回答已采纳
我很难将以下三个概念( RowTypeInfo、Row和DataStream )组合在一起。val mutationStream: DataStream[Mutation] = ...
// toRows returns an object of type org.apache
浏览 0提问于2018-02-09得票数 0
回答已采纳
我使用的是使用Java的Flink表API，其中我想将DataSet转换为DataStream .以下是我的代码：DataSet<MyClass>ds2=tableEnvironment.toDataSet(tab1, MyClass.class);
DataStreamMyClass> d=tableEnviro
浏览 4提问于2016-05-13得票数 4
KeyBy和GroupBy在Flink中的异同是什么？如果在仅表程序中使用表/SQL，那么GroupBy是否等同于KeyBy？
浏览 7提问于2020-09-04得票数 1
回答已采纳
{
DataStream<Tuple2<String, Integer>> counts = text
浏览 0提问于2019-11-13得票数 1
云服务器
对象存储
ICP备案
云点播
实时音视频
腾讯云开发者
Copyright © 2013 - 2026 Tencent Cloud. All Rights Reserved. 腾讯云 版权所有
深圳市腾讯计算机系统有限公司 ICP备案/许可证号：粤B2-20090059 
粤公网安备44030502008569号
腾讯云计算（北京）有限责任公司 京ICP证150476号 | 京ICP备11018762号