问题标签 [aggregator]

我正在尝试查询一个表并计算其他两个列的成对组合的列值的运行总和。

具体来说，给定下表：

我想查询一个计算每个桶/标签对的累积总和的列。

换句话说，我希望能够编写一个查询来选择存储桶、标签和第三列，该第三列是每个条目的“金额”列的累积总和，其标签与该行的标签匹配。例如，对于上面的示例，第三列应具有以下值：

换句话说，值 5 是 1 和 4 的总和（“A”标签的前两个数量），值 7 是 2 和 5 的总和（“B”标签的前两个数量），等等

我知道我可以通过执行以下操作来获得整个列的累积总和：

但我不确定如何按标签拆分。

任何帮助深表感谢！

我对 Camel (2.24.x) 比较陌生，我正在尝试在 Java DSL 中使用 AggregationStrategy，但有一些相关的问题/顾虑。另请注意，虽然我选择使用 Java DSL，但我也会接受 Spring Camel XML 格式的答案/解释，如果这是我能得到的最好的！

我正在尝试像“阀门”一样使用 AggregationStrategy，这意味着，收集/聚合入站交换，直到它确定它准备好将该聚合交换移动到路由的其余部分，但不允许任何交换/消息通过它，直到它下定决心。所以本质上，这个“阀门”（聚合器）是关闭的，直到它确定它已经充分聚合了 1 次以上的交换，然后一旦它做出这个决定，它就会“打开”。

要实现 AggregationStrategy 接口，准系统代码如下：

oldExchange 和 newExchange 分别对应什么？oldExchange 是否是我们正在聚合的消息，而 newExchange 只是我们可以（可选）附加到 oldExchange 的下一条消息？aggregate(...) 方法应该返回什么，oldExchange 还是 newExchange？或者是其他东西？！

然后，一旦我确定（在 aggregate(...) 方法内）是时候“打开阀门”了（例如，我们已经成功聚合 100%），我如何告诉 Camel 允许聚合交换通过到路线的其余部分？

提前感谢您提供全面的见解！

kubernetes官网建议聚合器最好配置不同的ca证书；凭据。于是，我按照官网的建议，重新生成了一个ca证书，并用这个ca签署了聚合器使用的证书。官网。然后启动api-server，但是启动失败。失败日志如下：

我所做的所有步骤如下：

第 1 步：生成证书

第二步：配置参数

vim /etc/kubernetes/apiserver

第三步：将启动参数添加到启动文件中

第四步：启动 kube - apiserver 启动失败，日志如上

嘿，任何人都可以向我解释为什么我需要使用流利的聚合器？以下是我一直在研究的基础设施

Fluentbit(Log Forwarder) --> Fluentd(Data Processing and Aggregator) --> Elasticsearch --> Kibana

抱歉没有显示图片，但您可以在这里找到我的意思：https ://medium.com/redbox-techblog/building-an-open-data-platform-logging-with-fluentd-and-elasticsearch- 4582de868398

好吧，也许这是一个奇怪的问题，但我仍然不明白我为什么需要它？当我只能使用fluentbit将日志直接转发到elasticsearch时，为什么我们必须需要聚合器？

有关更多信息，我将为几个节点使用大量 fluentbit（我认为大约 50 个节点）

如果你能帮助我，我很感激，对不起我的英语不好

谢谢

Spark 3.0 已弃用UserDefinedAggregateFunction，我试图使用Aggregator. 的基本用法Aggregator很简单，但是，我很难使用更通用的函数版本。

我将尝试用这个例子来解释我的问题，一个collect_set. 这不是我的实际情况，但更容易解释问题：

我更喜欢.toColumnvs .udf.register，但这不是重点。

问题： 我不能制作这个聚合器的通用版本，它只适用于整数。

我尝试过：

它因错误而崩溃：

我不能去CollectSetDemo[T]，万一我不能正常outputEncoder。此外，在使用 udaf 时，我只能使用 Spark 数据类型、列等。

我有一个数据库

现在我想根据状态为 ACTIVE 和 INACTIVE 的电子邮件找到该项目。我已经编写了查询来查找这样的重复项。

这会返回 tut@gmail.com 和 amrit@gmail.com 但我只想要 amrit@gmail.com 因为它在 db 中既是 ACTIVE 又是 INACTIVE。结果应该看起来像

我创建了一个 ref 集合，其中包含其他集合名称和文档 ID。

所以看起来像

集合 1 和集合 2 具有 id 123123 和 456734 的文档。在这种情况下，我想循环所有 ref 集合并使用 mongodb 聚合获取引用的集合数据。

我尝试使用 $lookup 但据说 $lookup 的 $from 必须是字符串文字，而不是变量。

无论如何在MongoDB中处理这种事务？

我正在尝试使用此示例创建自定义聚合器。但是在这里我可以看到我们只是基于从客户端传递的相同参数聚合两个下游服务响应。我的要求是我从客户端接收一个参数并将其传递给我的第一个下游服务，然后使用来自下游服务响应的数据作为下一个下游服务的参数。

例如：我传递了 Book id，它应该点击 Book Microservice 来获取图书信息，然后假设我们将 AuthorId 作为来自第一个下游服务的 BookInfo 响应中的模型属性之一，我使用这个 AuthorId 作为 Author 微服务的参数来获取作者信息。

我正在使用spring集成聚合器和MessageGroupStoreReaper，但不知何故错误没有到达全局errorChannel。

如果在“processInputChannel”后有任何异常（例如到期时的部分结果），那么异常不会到达全局“errorChannel”。

即使我尝试用轮询器将任务计划的工作替换为入站通道适配器（如@Gary所建议的那样），但它仍然不起作用：

</int:inbound-channel-adapter>

请建议

谢谢

我想用 Spark 做一些自定义 groupBy 聚合，这需要按顺序处理记录（时间戳），并且第 n 条记录的处理需要前（n-1）条记录的处理输出（听起来有点像流式传输任务？）。输入位于按日期分区的一大组文件中。

我目前的解决方案是实现一个 custom org.apache.spark.sql.expressions.Aggregator，它将所有输入记录增量插入缓冲区并在最后进行所有聚合。伪代码如下：

这个解决方案有效，但我对性能有很大的担忧，因为 reduce 阶段根本不会减少任何东西，并且所有记录都需要存储在内存中直到最后。我希望有更好的解决方案。

能否请你帮忙？谢谢。