1回答
我正在尝试在Flink SQL的跳跃窗口上做一个指数衰减的移动平均值。我需要有权访问窗口的边框之一，在下面的HOP_START： SELECT:66) 当它在聚合和之外工作时，它确实是未实现的。现在，问题是:我可以转换这个表达式并在聚合之外进行最后的处理，例如exp(x+y) = exp(x)*exp(y)；但是我坚持使用TIMESTAMPDIFF (which did wonders in我不知道如何才能让
浏览 115提问于2019-02-13得票数 0
1回答
翻转窗口触发时间
、、、、
我正在使用Flink来聚合来自kafka主题的数据。我使用的滚动窗口为1小时，时间特征设置为事件时间。我还使用AscendingTimestampExtractor，并根据json中的特定字段值为输入分配水印。env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
我们可以假设用于水印的JSON字段具有与系统时钟类似的时间值。如果我在下午3点25分部署作业，什么时候
浏览 3提问于2020-03-11得票数 3
回答已采纳
1回答
我在Amazon的服务(托管Flink集群)中运行了一个Flink应用程序。在这个应用程序中，我从一个名为keyBy userId的动态流中读取用户数据，然后聚合一些用户信息。在询问之后，我了解到Flink将在集群中的物理主机上拆分流的读取。Flink然后将传入的事件转发给将聚合器任务分配给对
浏览 4提问于2020-02-18得票数 0
回答已采纳
我的环境是：numpy version is............ 1.14.3我的熊猫数据框看起来像这样现在，如果我添加"Min“原语，我会得到： File "H:\ML\BlogExperiments\Python\SKLearn(就像Log10原语
浏览 16提问于2018-09-01得票数 1
回答已采纳
我是Flink的新手，我一直在遵循SocketWindowWordCount的例子。我正在使用Scala 2.11.8和Flink 1.3.2，并尝试在EMR上运行它，当我运行以下代码时，它抛出了错误：主类如下所示：
import org.apache.flink</
浏览 7提问于2017-10-27得票数 4
我的Flink工作经常与一个或另一个任务管理器进行OOM。我有足够的内存和存储空间来完成我的工作(2个JobManager/16 TaskManagers -每个都有15核和63 16内存)。有时作业在抛出OOM前4天运行，有时作业在2天内进入OOM。但与前几天相比，交通状况是一致的。
我收到一个建议，不要在流管道中通过对象，而是使用原语来减少洗牌开销和内存消耗。我工作的
浏览 3提问于2021-04-25得票数 1
这个问题因情况而异，你把我的问题标为重复的问题并不能完全澄清我的问题情况。Integer) enumeration.nextElement(); }}
为什么我们在枚举中写整数而不是整数
浏览 5提问于2016-05-29得票数 0
回答已采纳
我很肯定这个过程会是这样的：
在MetricReporter中聚合度量:所有被称为"
浏览 1提问于2019-06-26得票数 0
回答已采纳
1回答
我已经用Flink streaming做了一段时间的实验，使用了像Yahoo streaming benchmark：这样的基准测试，它应该给系统带来压力，但我从来没有达到令人满意的CPU利用率--事实上，使用所有可用的系统核心(并行度=节点*核心)和每个核心一个TaskManager插槽，它主要是低至约25%。最近，我开始使用Gelly，Flink的Graph API，使用一些提供的示例算法(例如Pagerank)，批处理数据集从数万到数亿个顶点。我
浏览 15提问于2018-01-18得票数 4
我有一个项目，建立了许多不同品种的海龟。每个品种都需要一只领头龟和一只跟随龟。我已经将这些作为全局变量分配给它们，因为它们在代码中出现了更多的下降。我需要将这些变量分配给品种中的海龟，然后创建一个从领导者到跟随者的链接。界面中有很多条件来决定有多少品种被创建，所以我不能根据海龟的数量来分配。我收到了一个错误(不是所有的时间)，“海龟无法链接到自己”，我认为当它们覆盖第一个set命令并将相同的海龟分配给这两个变量时，就会发生这种错误。有没有人知道我可以设
浏览 2提问于2018-03-12得票数 1
回答已采纳
我有一个特定的任务，使用Apache Flink和一些额外的逻辑将两个数据流连接到一个聚合中。事件数据流( Data of events )开始根据某些键在某个聚合或窗口中不断积累Type A;The记录的
浏览 6提问于2020-07-06得票数 0
回答已采纳
1回答
嗨，我正在尝试运行一个flink scala应用程序，它从kafka读取，应用一些查找转换，然后写到kafka。我在当地测试过，效果很好。但是，当我尝试使用原生kubernetes集成在集群上运行它时，我会看到下面这样的奇怪错误。
集群看起来也很好，因为我试图在集群上运行一个wordcount应用程序，它运行得很好。异常不清楚，堆栈跟踪也显示了任务管理器堆栈跟踪，因此不知道应用程序中的问题可能在哪里。这可能是序列化问题吗？是否有一种方法来调
浏览 2提问于2021-09-01得票数 0
回答已采纳
摄入卡夫卡的数据，比如50个分区(输入速率- 100,000毫希/秒)
将输出存储在Cassandra上我是否需要在google数据流中编写数据管道来执行flink上的步骤序列，还是有其他方法来执行实时流的步骤序列？假设我的每一次计算花费大约20毫秒，我如何更好地使用flink来设计它并获得更好的吞吐量
浏览 2提问于2016-06-10得票数 3
我已经实现了一个Flink处理器，它将事件聚合到会话中，然后将它们写入接收器。现在我想扩展它，这样我就可以获得每五分钟并发会话的数量。然后，我在Flink中执行以下操作，将事件聚合到会话中。trigger(SessionProcessingTimeTrigger.create())
.aggregate(new SessionAggregator()) 每个会话都是在处理具有<
浏览 12提问于2020-11-06得票数 0
回答已采纳
在我的scala代码中，我有一个Class[_ <: AnyVal]实例，例如从classOf[Int]获得的实例。当我尝试从这里(使用classOf[Int].getName)获得scala类型名称时，我希望看到"scala.Int"，但是我得到的是基本"scala.Int"类型的名称："int"
如何从我的Class变量中获得scala类型名，如果我处理的是java
浏览 3提问于2019-04-20得票数 1
回答已采纳
3回答
我要使用apache运行简单的worcount示例。使用$SPARK_HOME/jars中的本地jar文件，它正确运行，但使用maven依赖于它的错误： <groupId>org.apache.flink</groupId>
<artifactId>
浏览 19提问于2017-11-28得票数 0
回答已采纳
1回答
在阅读了flink的文档和搜索之后，我不能完全理解flink是如何处理窗口中的状态的。假设我有一个每小时滚动一次的窗口，其中包含一个聚合函数，该函数将msg累积到某个java pojo或scala case类中。该窗口的大小是与在一个小时内进入该窗口的事件的数量相关，还是仅仅与pojo/case类相关，因为im将事件累
浏览 49提问于2019-03-20得票数 5
回答已采纳
我不会假装完全了解这个mysql select中发生了什么，但我需要它的mssql等效项。它依赖于mysql ONLY_FULL_GROUP_BY配置被设置为默认值。Voorwerp v LEFT JOIN on v.voorwerpnummer=b.voorwerp 基本上，我希望我能得到与这个mysql查询在mssql中得到的完全相同的结果。但是，您不能像在mys
浏览 14提问于2020-05-05得票数 0
我正在尝试建立一个非常基本的flink工作。at org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getStreamGraph(StreamExecutionEnvironment.java:654)错误是由以下代码引起的：
val streamExecutionEnvironme
浏览 1提问于2019-03-04得票数 11
回答已采纳
云服务器
ICP备案
云直播
对象存储
实时音视频
腾讯云开发者
