数据回放

replay(inputTables, outputTables, [dateColumn], [timeColumn], [replayRate], [absoluteRate=true], [parallelLevel=1])

replay函数的作用是将若干表或数据源同时回放到相应的输出表中。用户需要指定输入的数据表或数据源、输出表、日期列、时间列、回放速度以及并行度。

• inputTables: 单个表或包含若干表或数据源（见replayDS介绍）的元组。
• outputTables: 可以是单个表或包含若干个表的元组，表示共享的流数据表对象，也可以是字符或字符串，表示共享流数据表的名称。
• dateColumn 与 timeColumn: 字符串，表示输入表的日期和时间列，若均不指定则默认第一列为 dateColumn。若有 dateColumn，则该列必须为分区列之一；若无 dateColumn，则必须指定 timeColumn，且其必须为分区列之一。若输入表中时间列同时包含日期和时间，需要将 dateColumn 和 timeColumn 设为同一列。回放时，系统将根据 dateColumn 和 timeColumn 的设定，决定回放的最小时间精度。在此时间精度下，同一时刻的数据将在相同批次输出。举例来说，若 timeColumn 最小时间精度为秒，则每一秒的数据在统一批次输出；若只设置了 dateColumn，那么同一天的所有数据会在一个批次输出。
• replayRate: 整数，表示数据回放速度。若 absoluteRate 为 true，replayRate 表示每秒钟回放的数据条数。由于回放时同一个时刻数据在同一批次输出，因此当 replayRate 小于一个批次的行数时，实际输出的速率会大于 replayRate。若 absoluteRate 为 false，依照数据中的时间戳加速 replayRate 倍回放。若 replayRate 未指定或为负，以最大速度回放。
• absoluteRate 是一个布尔值。默认值为 true，表示 replayRate 为每秒回放的记录数。若为 false，表示依照数据中的时间戳加速 replayRate 倍回放。
• parallelLevel: 整数，表示读取数据的并行度。当源数据单个分区相对内存较大，或者超过内存时，需要使用replayDS函数将源数据划分为若干个小的数据源，依次从磁盘中读取数据并回放。参数 parallelLevel 指定同时读取这些经过划分之后的小数据源的线程数，可提升数据读取速度。

replayDS(sqlObj, [dateColumn], [timeColumn], [timeRepartitionSchema])

replayDS函数可以将输入的 SQL 查询转化为数据源，结合replay函数使用。其作用是根据输入表的分区以及 timeRepartitionSchema，将原始的 SQL 查询按照时间顺序拆分成若干小的 SQL 查询。

• sqlObj: SQL 查询元代码，表示回放的数据，如 select * from loadTable("dfs://source", "source")。SQL 查询的表对象是 DFS 表，且至少有一个分区列为 DATE 类型。
• dateColumn: 字符串，表示日期列。若不指定，默认第一列为日期列。replayDS函数默认日期列是数据源的一个分区列，并根据分区信息将原始 SQL 查询拆分为多个查询。
• timeColumn: 字符串，表示时间列，配合 timeRepartitionSchema 使用。
• timeRepartitionSchema: TIME 或 NANOTIME 类型向量，如 08:00:00 .. 18:00:00。对 sqlObj 在每一个 dateColumn 分区中，在 timeColumn 维度上进一步拆分。

单内存表回放只需要设置输入表、输出表、日期列、时间列和回放速度即可。

replay(inputTable, outputTable, `date, `time, 10)

若数据表的行数过多，可使用replayDS函数将其划分为若干个小的数据源，再使用replay函数从磁盘中读取数据并回放，replayDS函数的返回值是一个向量，记录了划分出的多个 SQL 查询语句。replay内部实现使用了 pipeline 框架，取数据与输出分开执行。当replayDS函数的输出作为数据源时，多块数据可以并行读取，以避免输出线程等待的情况。此例中并行度设置为 2，表示有两个线程同时执行取数据的操作。

inputTable = loadTable("dfs://source", "source")
inputDS = replayDS(<select * from inputTable>, `date, `time, 08:00:00.000 + (1..10) * 3600000)
replay(inputDS, outputTable, `date, `time, 1000, true, 2)

如果replay函数是通过submitJob调用，可以使用getRecentJobs获取 jobId，然后用cancelJob取消回放。

getRecentJobs()
cancelJob(jobid)

如果replay函数是直接调用，可在另外一个 GUI session 中使用getConsoleJobs获取 jobId，然后使用cancelConsoleJob取消回放任务。

getConsoleJobs()
cancelConsoleJob(jobId)

本例中使用美国股市 2007 年 8 月 17 日的 level1 报价数据，执行replayDS函数进行数据回放，并通过 DolphinDB 内置的横截面聚合引擎计算 ETF 内在价值。数据存放在分布式数据库"dfs://TAQ"的 quotes 表，以下是 quotes 表的结构以及数据预览。

quotes = loadTable("dfs://TAQ", "quotes")
quotes.schema().colDefs;

select top 10 * from quotes where date=2007.08.17;

(1) 对要进行回放的数据进行划分。回放大量数据时，若将数据全部导入内存后再回放，可能导致内存不足。可先使用replayDS函数并指定参数 timeRepartitionSchema，将数据按照时间戳分为 60 个部分。

trs = cutPoints(09:30:00.000..16:00:00.000, 60)
rds = replayDS(<select * from quotes where date=2007.08.17>, `date, `time,  trs);

(2) 定义输出表 outQuotes 为流数据表。

sch = quotes.schema().colDefs
share streamTable(100:0, sch.name, sch.typeString) as outQuotes

(3) 定义 ETF 成分股票权重字典 weights 以及聚合函数 etfVal，用于计算 ETF 内在价值。为简化起见，本例使用了一个虚构的由 AAPL、IBM、MSFT、NTES、AMZN、GOOG 这 6 只股票组成的的 ETF。

defg etfVal(weights,sym, price) {
    return wsum(price, weights[sym])
}
weights = dict(STRING, DOUBLE)
weights[`AAPL] = 0.1
weights[`IBM] = 0.1
weights[`MSFT] = 0.1
weights[`NTES] = 0.1
weights[`AMZN] = 0.1
weights[`GOOG] = 0.5

(4) 创建流聚合引擎，并订阅数据回放的输出表 outQuotes。订阅 outQuotes 表时，我们指定过滤条件，只有 symbol 为 AAPL、IBM、MSFT、NTES、AMZN、GOOG 的数据才会发布到横截面聚合引擎，减少不必要的网络开销和数据传输。

setStreamTableFilterColumn(outQuotes, `symbol)
outputTable = table(1:0, `time`etf, [TIMESTAMP,DOUBLE])
tradesCrossAggregator=createCrossSectionalAggregator("etfvalue", <[etfVal{weights}(symbol, ofr)]>, outQuotes, outputTable, `symbol, `perBatch)
subscribeTable(tableName="outQuotes", actionName="tradesCrossAggregator", offset=-1, handler=append!{tradesCrossAggregator}, msgAsTable=true, filter=`AAPL`IBM`MSFT`NTES`AMZN`GOOG) 

(5) 开始回放，设定每秒回放 10 万条数据，聚合引擎则会实时地对回放的数据进行消费。

submitJob("replay_quotes", "replay_quotes_stream", replay, rds, outQuotes, `date, `time, 100000, true, 4)

(6) 查看 ETF 内在价值。

//查看outputTable表内前15行的数据,其中第一列时间为聚合计算发生的时间
select top 15 * from outputTable;