我想知道在哪些情况下Spark将作为UDAF函数的一部分执行合并操作。
动机:我在我的Spark项目中使用了许多窗口中的UDAF函数。通常我想回答这样一个问题:
在过去30天内,有多少次信用卡交易与当前交易发生在同一个国家?
窗口将从当前交易开始,但不包括它在计数中。它需要当前交易的值来确定在过去30天内要计算哪个国家。
当我在本地主节点上使用16个CPU和Spark 2.0.1版本运行它时,窗口中从未有任何合并,并且窗口中第一笔交易始终是当前交易。这正是我想要的。在不久的将来,我将在一个x100倍更大的数据集上运行我的代码,并在真正的分布式Spark集群上运行,想知道是否会发生合并。
问题:
动机:我在我的Spark项目中使用了许多窗口中的UDAF函数。通常我想回答这样一个问题:
在过去30天内,有多少次信用卡交易与当前交易发生在同一个国家?
窗口将从当前交易开始,但不包括它在计数中。它需要当前交易的值来确定在过去30天内要计算哪个国家。
val rollingWindow = Window
.partitionBy(partitionByColumn)
.orderBy(orderByColumn.desc)
.rangeBetween(0, windowSize)
df.withColumn(
outputColumnName,
customUDAF(inputColumn, orderByColumn).over(rollingWindow))
我编写了一个自定义UDAF来进行计数。我通常使用.orderBy(orderByColumn.desc)
,而由于.desc
的作用,在计算过程中当前交易将在窗口中排在第一位。
UDAF函数需要实现merge
函数,以便在并行计算中合并两个中间聚合缓冲区。如果发生任何合并,我的当前交易
可能对不同的缓冲区不同,导致UDAF的结果不正确。
我编写了一个UDAF函数来计算我的数据集中的合并次数,并仅保留窗口中的第一项交易与当前交易进行比较。
class FirstUDAF() extends UserDefinedAggregateFunction {
def inputSchema = new StructType().add("x", StringType)
.add("y", StringType)
def bufferSchema = new StructType()
.add("first", StringType)
.add("numMerge", IntegerType)
def dataType = new StructType()
.add("firstCode", StringType)
.add("numMerge", IntegerType)
def deterministic = true
def initialize(buffer: MutableAggregationBuffer) = {
buffer(0) = ""
buffer(1) = 1
}
def update(buffer: MutableAggregationBuffer, input: Row): Unit = {
if (buffer.getString(0) == "")
buffer(0) = input.getString(0)
}
def merge(buffer1: MutableAggregationBuffer, buffer2: Row) = {
buffer1(1) = buffer1.getInt(1) + buffer2.getInt(1)
}
def evaluate(buffer: Row) = buffer
}
当我在本地主节点上使用16个CPU和Spark 2.0.1版本运行它时,窗口中从未有任何合并,并且窗口中第一笔交易始终是当前交易。这正是我想要的。在不久的将来,我将在一个x100倍更大的数据集上运行我的代码,并在真正的分布式Spark集群上运行,想知道是否会发生合并。
问题:
- UDAFF什么时候会发生合并?
- 带OrderBy的窗口是否会发生合并?
- 是否可以告诉Spark不要进行合并?