问题标签 [user-defined-aggregate]

我是 Cassandra 的新手，在本地机器上的 Cassandra 3 节点集群上运行用户定义的聚合。问题是，当我在较小的数据集上运行此聚合时，结果很好并且符合预期。

但是当数据太大时，查询失败并出现错误 -

OperationTimedOut: errors={'127.0.0.1': '客户端请求超时。见 Session.execute_async'}, last_host=127.0.0.1

我发现了与我的问题类似的以下问题，但没有得到解答。查找其他问题的链接 -

如何为大型用户定义的聚合查询设置超时和节流率

Cassandra CQLSH OperationTimedOut 错误=客户端请求超时。请参阅 Session.execute[_async](超时)

我已经修改了 cassandra.yaml 并且时间限制是 -

但这对我没有帮助。请指导这些时间的正确配置是什么，以便在没有查询超时的情况下在大型数据集上运行相同的查询。

聚合代码 -

谢谢并恭祝安康，

维巴夫

PS - 如果有人选择否决这个问题，请在评论中提及相同的原因。

我从我的 mongodb 中得到一些数据，如下所示：

没有必要知道里面是什么。

我写了一个这样的 UserDefinedAggregateFunction 因为我想分组查看：

我的输出应该是这样的：

这些值是在上面的更新/合并函数中计算出来的，但这是有效的，没有必要让你看到它。

然后我像这样使用它：

当我执行 df.show() 时，它总是给我一个 IndexOutOfBoundException。我知道这是懒惰的评估，这就是我在 df.show() 中收到错误的原因。

据我所见，它可以执行第一组并结束评估功能。但在那之后我得到一个 IndexOutOfBoundException ......

此外，当我更改 dataType 并将 Function 评估为：

输出如下所示：

架构如下所示：

而且我无法将其转换为我想要的形式。

因为第二种方法有效，我想我在第一种方法中弄乱了 DataType 的东西，但我不知道如何修复它......

我的问题有很多介绍：

我怎样才能得到我想要的输出？我真的不在乎这两种方法中的哪一种（首先具有多个输出列或可以转换为我想要的形式的数组），只要它是有效的。

谢谢你的帮助

我正在尝试创建一个按天计算和求和值的函数（以便稍后计算平均值）。我做到了这一点：

这失败了，因为 tupleValue 在每个新的一天都为空，但尚未在地图中。如何在 UDA 中实例化元组值？

下午，

我正在编写一个自定义中值函数（不​​看现有的解决方案，我喜欢挑战），经过大量的摆弄，我大部分时间都在那里。但是，我不能传入包含空值的列。我在 c# 代码中处理这个问题，但它似乎在它到达那里之前被 SQL 停止了。

你得到这个错误...

C＃：

SQL：

关于我缺少什么设置或代码的任何想法都将允许这样做。我几乎只是希望它忽略空值。

SQL版本是SQL2008 R2 btw

我正在尝试在 Spark（2.0.1，Scala 2.11）上创建一个 UDAF，如下所示。这本质上是聚合元组并输出Map

但是当我编译它时，我看到以下错误：

我究竟做错了什么？

编辑：对于将其标记为Spark UDAF 副本的人 - 使用泛型作为输入类型？- 这不是该问题的重复，因为该问题不处理Map数据类型。对于使用 Map 数据类型所面临的问题，上面的代码非常具体和完整。

我想知道在什么情况下 Spark 将作为 UDAF 功能的一部分执行合并。

动机： 我在我的 Spark 项目中的一个窗口上使用了很多 UDAF 函数。我经常想回答这样的问题：

在 30 天的窗口内，与当前交易在同一国家/地区进行了多少次信用卡交易？

该窗口将从当前事务开始，但不会将其包括在计数中。它需要当前交易的价值才能知道过去 30 天内要计算哪个国家/地区。

我写了我的 customUDAF 来进行计数。我总是使用.orderBy(orderByColumn.desc)并感谢.desc当前交易在计算过程中首先出现在窗口中。

UDAF 函数需要实现merge在并行计算中合并两个中间聚合缓冲区的函数。如果发生任何合并，current transaction不同缓冲区的 my 可能不一样，UDAF 的结果将不正确。

我编写了一个 UDAF 函数，它计算我的数据集上的合并次数，并只保留窗口中的第一个事务以与当前事务进行比较。

当我在具有 16 个 cpu 的本地主机上使用 spark 2.0.1 运行它时，从来没有任何合并，并且窗口中的第一个事务始终是当前事务。这就是我要的。在不久的将来，我将在 x100 更大的数据集和真正的分布式 Spark 集群上运行我的代码，并想知道那里是否可以发生合并。

问题：

• UDAF 在哪些情况/条件下进行合并？
• 带有 orderBy 的 Windows 是否曾经进行过合并？
• 是否可以告诉 Spark 不要进行合并？

我一直在尝试在 SQL Server 中创建自定义聚合函数。我终于让它工作了（用 C# 编写了程序集）。

目标是连接组内的所有字符串，但结果是空字符串。

这是我在 C# 中的课程：

这是 T-SQL 部分：

我正在创建一个需要附加参数的用户定义聚合函数。更准确地说，它是一个累积（又名窗口）最小值，它将定义窗口的时间间隔作为第二个参数。由于聚合函数对我的用户定义的数据类型进行操作，我已经传达了一个虚拟示例，该示例计算列的最后 n 个值的平均值。我知道我可以在 PostgreSQL 中解决这个虚拟问题，但该示例的目的只是为了突出我的问题。

我收到以下错误

如何告诉 PostgreSQL 我的最终函数也需要直接参数？我正在研究 PostgreSQL 10.1。我知道根据文档，直接参数只允许用于有序集聚合，但我还需要一个用于“普通”聚合的直接参数。

我正在尝试使用 Spark UDAF 将两个现有列汇总为一个新列。大多数关于 Spark UDAF 的教程都使用索引来获取输入行的每一列中的值。像这样：

，在我的更新方法 ( override def update(buffer: MutableAggregationBuffer, input: Row): Unit) 中使用。它也适用于我的情况。但是我想使用该列的字段名称来获取该值。像这样：

，其中 ColumnNames.BehaviorType 是对象中定义的 String 对象：

这一次它不起作用。我得到以下异常：

java.lang.IllegalArgumentException：字段“BehaviorType”不存在。在 org.apache.spark.sql.types.StructType$$anonfun$fieldIndex$1.apply(StructType.scala:292) ... 在 org.apache.spark.sql.Row$class.getAs(Row.scala:333 ) 在 org.apache.spark.sql.catalyst.expressions.GenericRow.getAs(rows.scala:165) 在 com.recsys.UserBehaviorRecordsUDAF.update(UserBehaviorRecordsUDAF.scala:44)

部分相关代码段：</p>

这是我的 UDAF 的一部分：

以下是调用UDAF的部分代码：

输入行模式中的字段名称似乎没有传递到 UDAF：

我的代码有什么问题？

我一直认为 Spark 不允许定义 User-Defined-Window-Functions。我刚刚从这里（ https://docs.databricks.com/spark/latest/spark-sql/udaf-scala.html ）测试了“几何平均值”UDAF示例作为窗口函数，它似乎工作得很好，例如：

我从未见过 spark 文档谈论使用 UDAF 作为窗口函数。这是否允许，即结果是否正确？顺便说一下，我正在使用 spark 2.1

编辑：

令我困惑的是，在标准聚合中（即后跟 a groupBy），数据总是添加到缓冲区中，即它们总是会增长，不会缩小。使用窗口函数（尤其是rowsBetween()与 我认为窗口函数可以随着状态的顺序移动。所以我认为必须有类似“删除”的方法来实现