mr

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

mr is a general-purpose DolphinDB function for performing MapReduce computations. When data is distributed across multiple data sources, you can first compute each data source separately, then merge the partial results, and finally apply any additional processing you need.

ds: A list of data sources. This parameter must be a tuple. Typically, each element in the tuple is a data source object. Even if there is only one data source, you must still set it as a tuple.

An element can also be a tuple of aligned data source objects. If ds contains inner tuples, the data sources in each inner tuple must come from distributed tables in the same database, and every outer element must expand to the same number of data sources. Use multiTableRepartitionDS to generate aligned data sources for multiple tables.

mapFunc: The computation function applied to each outer element of ds. If the element is a single data source object, mapFunc takes one parameter: the data object extracted from the data source and made available for computation. To pass additional fixed arguments, use partial application to bind them first. The map function is called once for each outer element of ds. It returns either a regular object (scalar, pair, vector, matrix, table, set, or dictionary) or a tuple containing multiple regular objects.

If an element of ds is an inner tuple, mr materializes its data sources and expands them into separate arguments. The number of unbound arguments of mapFunc must equal the number of data sources in the inner tuple. For example, if an inner tuple contains data sources from two tables, mapFunc can be defined as def(t1, t2) { ... }. To pass additional fixed arguments, use partial application to bind them first.

reduceFunc: Optional. A binary function used to merge two return values. These two return values may both come from the map function, or one of them may be the return value from the previous reduce function call. The system continues merging the result from the previous step with the return value of the next map function until it has merged the return values from all map functions.

finalFunc: Optional. A function that generates the final output. It takes only one parameter: the output of the last reduce function call. If you specify a final function but do not specify a reduce function, the system first combines all map function results into a tuple, and then calls the final function with that tuple as input.

parallel: Optional. Specifies whether to execute the map function in parallel. The default value is true. In general, enabling parallel execution can improve computation speed, but you may want to set this parameter to false in the following cases:

• A single map computation uses a very large amount of memory.
• Running multiple threads concurrently may cause thread-safety issues. For example, errors may occur if multiple threads write to the same file at the same time.

• If you specify finalFunc, mr returns the result of finalFunc.
• If you do not specify finalFunc but do specify reduceFunc, mr returns the final result of reduceFunc.
• If neither is specified, mr returns the aggregated results of all mapFunc calls as a tuple.

Example 1: Compute the global mean and variance across multiple partitions.

// Create a catalog and switch to it
createCatalog("demo")
go
use catalog demo
// Create a DFS database and a partitioned table in the current 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

// Insert test data
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)

// Use sqlDS to convert the query into a list of data sources
ds = sqlDS(<select * from mr_demo.pt>)


// map: Computes local statistics for each partition
def statsMap(table){
    x = table.value // Get the values from the value column of the table
    return [x.size(), sum(x), sum(pow(x, 2.0))] // Compute the row count, the sum of the values, and the sum of squares of the values
}

// reduce: Adds two local statistics item by item
def statsReduce(x, y){
    return [x[0] + y[0], x[1] + y[1], x[2] + y[2]]
}

// final: Computes the global mean and variance from the aggregated statistics
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)
}

// Call mr
result = mr(ds, statsMap, statsReduce, statsFinal)

// View the result
result

The output is as follows:

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

Example 3: Compute least-squares linear regression across multiple data sources.

Linear regression predicts a dependent variable from one or more independent variables. Let the dependent variable be y and the independent-variable matrix be X. The goal of ordinary least squares is to find the regression coefficient beta, computed as follows:

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

Therefore, the key to completing the regression is to first obtain X^T X and X^T y for the entire dataset. When data is distributed across multiple data sources, you can first compute the local Xi^T Xi and Xi^T yi on each data source, and then aggregate them into the global result. Sample code:

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 performs the local computation for a single data source. Its inputs are:
  • table: The data table corresponding to the current data source.
  • yColName: The name of the dependent-variable column.
  • xColNames: Independent variable column names.
  • intercept: Specifies whether to include an intercept term.

  This function first constructs the local independent variable matrix Xi, then returns the local statistics for the current data source: Xi^T Xi and Xi^T yi.

• reduce: Uses + as the reduce function to add the results from individual data sources item by item.
  • For example, if two partitions return (X1^T X1, X1^T y1) and (X2^T X2, X2^T y2), the reduce step produces (X1^T X1 + X2^T X2, X1^T y1 + X2^T y2).
  • It then merges this result with the next partition result and ultimately produces the global (X^T X, X^T y).
• final: myOLSFinal processes the aggregated result from the reduce phase and computes the regression coefficients using the least squares formula.
  • xtx: The aggregated X^T X.
  • xty: The aggregated X^T y.
  • xtx.inv(): The inverse of the matrix xtx.
  • xtx.inv().dot(xty): Matrix multiplication, corresponding to the formula (X^T X)^(-1) X^T y.
• myOLSEx wraps the mr function and runs the computation through the map → reduce → final workflow.