也许这种方法是不正确的。 我想我应该使用groupBy。另一件事是，我可能需要第一次使用流来获得组的键列表(或者可能找到一种方法来使用它一次，但会产生两个结果)。此外，我想我应该提取parentTconst来使用它们作为分组关键字： def getParentTConst: Flow[Episode, Set[String], _] = Flow[Episode].fold(Set.empty[String]) {
(right,
浏览 7提问于2019-06-13得票数 2
在第270行，使用HashMap对流文件进行分组，然后迭代entrySet()以将流文件添加到readyBins。 ? 如果没有设置Correlation Attribute Name，而Merge Strategy设置为Bin-Packing Algorithm，那么MergeContent如何保持从队列中拉出流文件的顺序
浏览 10提问于2019-09-18得票数 0
允许按键对元素进行分组，并在特定事件或时间间隔(即，来自RxJS的描述)之后将其移除。
正如我所看到的，apache-flink没有这样的开箱即用功能。谁能解释一下如何在apache-flink中实现这样的功能？
浏览 0提问于2015-11-25得票数 1
假设我有一个带有键和时间戳的流。我想在每个窗口(滑动窗口)中创建这些关键点的笛卡尔乘积。如果我有键1,2,3,4，并且我将并行度设置为2，我想用以下方式对它们进行“分组”：1 - 3 2 - 4并且我希望在每个窗口中按组处理元素因此，假设上面的元素(1,2,3,4)基于它们的时间戳在同一个窗口中。在最简单
浏览 7提问于2017-02-10得票数 0
2回答
我有一个发射Either[String, MyClass]的消息来源。我想用批MyClass调用外部服务，然后继续向下游使用Either[String, ExternalServiceResponse]，这就是为什么我需要对流元素进行分组。如果流只发出MyClass元素，那么很容易调用grouped。(10) // Seq[MyClass]
.map(callExternalService(_))
浏览 6提问于2020-03-04得票数 1
我正在处理一个按键分组的文件中的数据流。我已经创建了一个类，该类具有一个apply方法，该方法可用于按名为KeyChangesT，K的键对流进行拆分。在处理子流的第一项之前，我需要从DB检索一些数据。完成每个子流之后，我需要向队列发出一条消息。在标准scala序列中，我会这样做：
val groups: Map[Key, Seq[Value]] = stream.
浏览 5提问于2016-03-31得票数 2
回答已采纳
2回答
我开始学习Java，我想知道是否可能只查看流的第一个元素而不检索它。例如，我有多个流，每个流都有按非递减顺序排序的整数，我希望得到所有整数的排序列表，所以我正在考虑使用一个按非递减顺序排序的PrioirtyQueue<Stream>。但是，为了让PrioirtyQueue<Stream>对流进行排序，我需要为流传递一个比较器，以便通过它们的第一个元素
浏览 6提问于2022-05-16得票数 1
回答已采纳
1回答
我对流图方法有点困惑。有一个简单的源代码可以运行： List<String> words = Arrays.asList("hellolove", "you");map()签名:流映射在我的选项中，
浏览 6提问于2022-02-27得票数 -1
window(GlobalWindows.create()) .sum(2);
有人能解释一下为什么在这个例子中使用GlobalWindow的一个实例(在GlobalWindows#assignWindows中创建)吗？对于任何传入事件ID，都应该创建自己的窗口，即Window(a)表示a事件，Window(b)表示b事件，等等，因为据我所知，Flink使用窗口实例来关联对应的事件，例如，所
浏览 4提问于2016-03-30得票数 2
回答已采纳
我怎样才能发射到多个流从同一螺栓在风暴三叉树？将流拆分为多个流：public void declareOutputFields(final OutputFieldsDeclarer");
builder.setBolt("type1Bolt", type1Bolt).shuffleGro
浏览 3提问于2015-11-21得票数 1
1回答
我收到了以下随机顺序的信息：
在我的应用程序中，我需要计算具有相同id的最后两个记录，然后立即发送响应。
消息(timestamp=.使用相同的索引字段存储和计算最后两条记录的最佳方法是什么？你能告诉我一些提示吗？
浏览 0提问于2019-07-09得票数 0
2回答
我无法理解kafka流中的groupBy/groupById和窗口的概念。我的目标是在一段时间内(例如5秒)对流数据进行聚合。我的流数据看起来类似于：{"value":10,"time":1533875667511}
{"value":8,"time
浏览 0提问于2018-08-10得票数 7
1回答
我在网上遇到了一些文章，描述了通过流和块密码组合而成的密码。(参见Sandy：https://eprint.iacr.org/2008/473.pdf的这篇2008年论文)简而言之，概念是使用流密码的输出来添加白化和/或更新分组密码的圆键。对流密码部分的标准攻击应被分组密码所阻止，而对分组密码的攻
浏览 0提问于2016-12-28得票数 3
我想知道streams (或Collector)中是否已经实现了一个功能，它首先按属性对流进行分组，然后返回列表中按另一个属性排序的第一个元素。例如，下面的代码尝试使用第一个属性对对象流进行分组，然后希望收集具有第二个属性的最高值的对象。class MyClass{ String att2;现在，我想用流myClassStrea
浏览 58提问于2020-03-15得票数 5
回答已采纳
在Google /Sub 中，关于拉出交付中的负载平衡问题，可以这样说：
我担心的是最后一个阶段。我能决定如何划分这个主题吗？换句话说，我可以决定子集的分组方式吗？例如，在Kinesis服务中，我可以通过用户id来确定流的分区键，因此，使用<
浏览 2提问于2017-04-11得票数 4
回答已采纳
我正在尝试编写一个数据流作业，它需要处理存储中的日志并将它们写入不同的BigQuery表中。要使用哪些输出表取决于日志中的记录。因此，我对日志进行了一些处理，并根据日志中的值生成了一个键。在此之后，我将日志分组在密钥上。我需要将分组在同一个键上的所有日志写入到一个表中。我正在
浏览 2提问于2019-07-09得票数 0
我正在开发一个网络游戏，大体如下：其目的是侦听来自特定客户端的消息，并在服务器上对其进行处理。每个Run方法都是某种无限循环，其start命令(阻塞命令)：然后通过在接收到的命令上切换外壳结构来进行处理。处理完成后，循环会使代码返回到阻塞的m_In.readInt()
来等待另一个命令的到来。我的问题是:我的网
浏览 0提问于2011-10-18得票数 0
回答已采纳
4回答
我希望根据Stream的内容将单个Stream拆分为Streams。生成的Stream应该包含原始流的部分数据。我的实际应用程序要复杂得多(在时间间隔的列表中对日志行进行分组)，但我的问题是如何处理流，因此我在这里询问一个简化的示例。因为它意味着对每个输入值进行两次评估，这是可以接受的，但我更希望避免这样做。解决方案的<
浏览 4提问于2015-06-24得票数 8
回答已采纳
1回答
我想读取一个CSV文件从SFTP的~8 8gb的。什么是最好的方式来流的CSV文件也做并行转换？我们可以并行地做流媒体和转换吗？如果是这样的话，我们如何确保转换所需的数据在流中可用呢？注意:我使用的是Mule4.1.2运行时。任何关于这方面的帮助都会很好。提前感谢！
浏览 16提问于2018-05-29得票数 0
2回答
在Java中使用流时，是否可以使用方法引用来进行算术加法(而不是lambda)？目前，我正在使用stream.reduce(0, (x, y) -> x + y) (对流的内容进行求和)，但如果存在类似stream.reduce(0, Math::add)的内容，则希望使用它。在更一般但类似的说明中，Java中的每一种方法/操作都
浏览 1提问于2020-01-20得票数 5
回答已采纳
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
