mr

mr(ds, mapFunc, [reduceFunc], [finalFunc], [parallel=true])

mr 是 DolphinDB 中执行 MapReduce 计算的通用函数。 当数据分布在多个数据源上时，可以先对每个数据源分别计算，再把各部分结果合并，最后按需要做一次统一处理。

ds 数据源列表，必须是一个元组。通常，元组的每个元素都是数据源对象。 即使只有一个数据源，也需要写成元组形式。

此外，元组的元素也可以是由多个对齐的数据源对象组成的子元组。对于包含子元组的 ds，同一子元组中的数据源必须来自同一数据库的分布式表，且各外层元素展开后的数据源个数必须一致。可使用 multiTableRepartitionDS 获取按分区对齐的多表数据源。

mapFunc 针对 ds 的每个外层元素执行的计算函数（即 map 函数）。当元素为单个数据源对象时，它接收一个参数，即“从当前数据源中提取出来、可供计算的数据对象”。 若还需传入其他固定参数，可以使用部分应用预先绑定这些参数。map 函数调用次数等于 ds 的外层元素个数，其返回值为一个常规对象（标量，数据对，向量，矩阵，表，集合或字典）或一个元组（包含多个常规对象）。

若 ds 的元素是子元组，mr 会物化并展开其中的数据源，mapFunc 的未绑定参数个数必须与子元组长度一致。例如，子元组包含两张表的数据源时，可将 mapFunc 定义为 def(t1, t2) { ... }。若还需传入固定参数，可使用部分应用绑定这些参数。

reduceFunc 可选，一个二元函数（即 reduce 函数），用于合并两个返回值。 这里的两个返回值，可能都来自 map 函数，也可能其中一个是前一个 reduce 函数的返回值。 系统会把上一步的合并结果继续与下一个 map 函数的返回值合并，直到所有 map 函数的返回值都被合并完。

finalFunc 可选，用于生成最终输出的函数。 它只接收一个参数，即最后一次调用 reduce 函数的输出。 如果没有指定 reduce 函数但指定了 final 函数，系统先将所有 map 函数的结果合并成一个元组，然后将元组作为输入来调用 final 函数。

parallel 可选，是否并行执行 map 函数，默认值为 true。 一般情况下，开启并行可以提升计算速度，但在以下情况可以考虑设为 false：

• 单次 map 计算非常占内存；
• 多个线程并发执行时可能引发线程安全问题； 例如，如果多个线程同时写入同一个文件，则可能会发生错误。

• 如果指定了 finalFunc，返回 finalFunc 的结果；
• 如果没有指定 finalFunc，但指定了 reduceFunc，返回 reduceFunc 的最终结果；
• 如果两者都没有指定，返回所有 mapFunc 的汇总结果，形式为一个元组。

例 1：计算多个分区中的全局均值和方差。

// 创建 catalog，并切换到该 catalog
createCatalog("demo")
go
use catalog demo
// 在当前 catalog 下创建分布式数据库和分区表
create database mr_demo partitioned by VALUE(2024.01.01 2024.01.02 2024.01.03), engine='OLAP'
go
create table mr_demo.pt (
    date DATE,
    id INT,
    value DOUBLE
)
partitioned by date

// 写入测试数据
data = table(
    2024.01.01 2024.01.01 2024.01.01 2024.01.02 2024.01.02 2024.01.03 2024.01.03 as date,
    1 2 3 4 5 6 7 as id,
    10.0 20.0 30.0 40.0 50.0 60.0 70.0 as value
)
demo.mr_demo.pt.append!(data)


// 用 sqlDS 把查询转换成数据源列表
ds = sqlDS(<select * from mr_demo.pt>)


// map：在每个分区中计算局部统计量
def statsMap(table){
    x = table.value // 取表格 value 列的值
    return [x.size(), sum(x), sum(pow(x, 2.0))] // 计算行数、值的总和、值的平方和
}

// reduce：把两个局部统计量逐项相加
def statsReduce(x, y){
    return [x[0] + y[0], x[1] + y[1], x[2] + y[2]]
}

// final：根据汇总统计量计算全局均值和方差
def statsFinal(result){
    count = result[0]
    totalSum = result[1]
    totalSquares = result[2]
    avg = totalSum / count
    variance = totalSquares / count - avg * avg
    return table(avg as mean, variance as variance)
}

// 调用 mr
result = mr(ds, statsMap, statsReduce, statsFinal)

// 查看结果
result

输出如下：

create table mr_demo.orders (
    TradeDate DATE,
    SecurityID SYMBOL,
    TradeTime TIME,
    OrderNO LONG,
    OrderQty INT
)
partitioned by TradeDate
go

create table mr_demo.trades (
    TradeDate DATE,
    SecurityID SYMBOL,
    TradeTime TIME,
    TradePrice DOUBLE,
    OfferApplSeqNum LONG
)
partitioned by TradeDate
go

orderData = table(
    2024.01.01 2024.01.01 2024.01.01 2024.01.02 2024.01.02 2024.01.02 as TradeDate,
    `AAPL`AAPL`MSFT`AAPL`AAPL`MSFT as SecurityID,
    09:30:00.000 09:31:00.000 09:32:00.000 09:30:00.000 09:31:00.000 09:32:00.000 as TradeTime,
    1001 1002 1001 2001 2002 2001 as OrderNO,
    100 300 200 150 150 400 as OrderQty
)
tradeData = table(
    2024.01.01 2024.01.01 2024.01.01 2024.01.02 2024.01.02 2024.01.02 as TradeDate,
    `AAPL`AAPL`MSFT`AAPL`AAPL`MSFT as SecurityID,
    09:30:00.300 09:31:00.700 09:32:00.200 09:30:00.100 09:31:00.400 09:32:00.100 as TradeTime,
    0.0 0.0 0.0 0.0 0.0 12.0 as TradePrice,
    1001 1002 1001 2001 2002 2001 as OfferApplSeqNum
)
demo.mr_demo.orders.append!(orderData)
demo.mr_demo.trades.append!(tradeData)

days = 2024.01.01..2024.01.02
ds = multiTableRepartitionDS(queries=[
    <select * from mr_demo.orders where TradeDate in days>,
    <select * from mr_demo.trades where TradeDate in days>
])

def calcCancelRatio(orderTB, tradeTB){
    startTime = 09:30:00.000
    endTime = 14:57:00.000
    cancelTB = select TradeDate, SecurityID, TradeTime as CancelTime, OfferApplSeqNum as OrderNO
        from tradeTB
        where TradeTime between startTime and endTime, TradePrice=0
    joinedTB = lj(orderTB, cancelTB, `TradeDate`SecurityID`OrderNO)
    return select TradeDate, SecurityID,
            1.0 * sum(iif(isNull(CancelTime), 0,
                iif((CancelTime-TradeTime>=0) and (CancelTime-TradeTime<500), OrderQty, 0))) /
                sum(OrderQty) as cancelRatio
        from joinedTB
        where TradeTime between startTime and endTime
        group by TradeDate, SecurityID
}

result = mr(ds=ds, mapFunc=calcCancelRatio, reduceFunc=unionAll)
result.sortBy!(`TradeDate`SecurityID)
result

例 3：在多个数据源上计算最小二乘线性回归的结果。

线性回归用于根据一个或多个自变量预测因变量。 若记因变量为 y，自变量矩阵为 X，则最小二乘法的目标是求出回归系数 beta，计算公式如下：

beta = (X^T X)^(-1) X^T y

因此，完成回归计算的关键在于先得到整个数据集的 X^T X 和 X^T y。 当数据分布在多个数据源上时，可以先在每个数据源上计算局部的 Xi^T Xi 和 Xi^T yi，再将它们汇总为全局结果。 示例代码：

def myOLSMap(table, yColName, xColNames, intercept){
  if(intercept)
      x = matrix(take(1.0, table.rows()), table[xColNames])
  else
      x = matrix(table[xColNames])
  xt = x.transpose();
  return xt.dot(x), xt.dot(table[yColName])
}

def myOLSFinal(result){
  xtx = result[0]
  xty = result[1]
  return xtx.inv().dot(xty)[0]
}

def myOLSEx(ds, yColName, xColNames, intercept){
  return mr(ds, myOLSMap{, yColName, xColNames, intercept}, +, myOLSFinal)
}

• map：myOLSMap 用于对单个数据源执行局部计算。 其输入包括：
  • table：当前数据源对应的数据表
  • yColName：因变量列名
  • xColNames：自变量列名
  • intercept：指定是否包含截距项

  该函数首先构造局部自变量矩阵 Xi，然后返回当前数据源的局部统计量：Xi^T Xi 和 Xi^T yi。

• reduce：+ 作为 reduce 函数，将各单个数据源的计算结果逐项相加。
  • 例如两个分区分别返回：(X1^T X1, X1^T y1) 和 (X2^T X2, X2^T y2)，那么 reduce 后得到 (X1^T X1 + X2^T X2, X1^T y1 + X2^T y2)；
  • 继续和下一个分区结果合并后，最终得到全局的 (X^T X, X^T y)。
• final：myOLSFinal 用于处理 reduce 阶段的汇总结果，即根据最小二乘公式求出回归系数。
  • xtx 表示汇总后的 X^T X；
  • xty 表示汇总后的 X^T y；
  • xtx.inv() 表示对矩阵 xtx 求逆；
  • xtx.inv().dot(xty) 表示做矩阵乘法，对应公式 (X^T X)^(-1) X^T y。
• myOLSEx 用于封装 mr 函数，执行 map → reduce → final 的计算逻辑。