利用 DolphinDB 高效清洗数据

常规情况下会怎么做呢？

通常我们会依次遍历数据集，对数据集的每个分片处理，然后汇总输出。

具体来说，分为如下几个步骤：

(1) 按交易日、股票 ID 将原始数据进行分片；

data = [cut1, cut2, ... , cutN]

(2) 依次遍历每个分片 cutK，使用 step1（代码第8行）至 stepM（代码第19行），共 M 个步骤来完成每个分片的清洗转化，并存储至内存对象 tradingdf；

(3) 将转化完成之后的内存对象 tradingdf 存储至 DFS 表。

代码如下：

def genDataV1(date1, dateN){
    tradeSrc = loadTable("dfs://originData", "trade")
    tradeTgt = loadTable("dfs://formatData", "trade")
    for (aDate in date1..dateN){
        tradeSecurityID = (exec distinct(securityID) from tradeSrc where tradingdate = aDate).shuffle()
        for (m in tradeSecurityID){		
		    tradingdf = select  * from tradeSrc where securityID = m and tradingdate = aDate    
		    tradingdf["symbol"] = m + "SZ"        
		    //print("stock " + m + ",date is " + aDate + ",tradingdf size " + tradingdf.size())  
		    tradingdf["buysellflag"] =iif(tradingdf["sellorderid"] > tradingdf["buyorderid"],"S", "B")
		    tradingdf["tradeamount"] = tradingdf["tradevolume"] * tradingdf["tradeprice"]
		    tradingdf = tradingdf[(tradingdf["tradetype"] == "0") || (tradingdf["tradetype"] == "F")]
		    tradingdf = select symbol,tradingdate, tradingtime, recid, tradeprice, tradevolume, tradeamount, buyorderid, sellorderid, buysellflag, unix from tradingdf
		    tradingdf = select * from tradingdf order by symbol, tradingtime, recid     
		    tradingdf.replaceColumn!("tradingdate", toIntDate(::date(tradingdf.tradingDate)))            
		    tradingtime = string(exec tradingtime from tradingdf)
		    tradingdf.replaceColumn!(`tradingtime, tradingtime)
		    unix = long(exec unix from tradingdf)
		    tradingdf.replaceColumn!(`unix, unix)                                             
		    tradeTgt.append!(tradingdf)	      		
        }
	}
}

通常我们使用 Python、MySQL、Java、中间件（如 Kettle）作为清洗数据的 ETL 工具时，受限于它们的单点性能。按上述代码执行20个交易日的数据需要4.5小时，处理速度十分缓慢。

回顾上述常规处理方式可以看到，优化前的代码效率较低，主要有以下几个原因：

(1) 双重 for 循环

常规的代码按股票、日期执行了双重循环，时间复杂度为：

t = O(N) * O(M) * t0 = O(MN) * t0

N: 交易日
M: 股票数量
t0: 最内层处理逻辑的耗时

指定日期和股票，测试最内层 for 循环执行一次的耗时为400 ms 左右。整体代码执行预估耗时为：

t ~= 20 * 0.4 * 3000 = 6.7 小时

(2) 数据访问次数太多

由上述代码可以看出，从 step1 (代码第8行)到 stepM（代码第19行）对同一个数据集操作了 M 次，这就导致代码执行起来速度较慢。而有一些操作可以同时处理，例如过滤数据与排序可以在一次操作中完成。

(3) 单点计算

上例中从 tradingdf 赋值语句开始：

tradingdf=select * from loadTable("dfs://test", 'szl2_stock_trade_daily') where symbol = m and tradingDate = date

之后的代码仅在单个节点进行计算，没有发挥出 DolphinDB 分布式、高并发计算的优势，从而当数据量不断增长，性能也就有所下降。

在 DolphinDB 中，可以通过分区剪枝、列存、索引等技术手段来降低问题空间复杂度。

• 分区剪枝：将时序数据进行分区（一般选择时间字段），当使用分区字段作为 where 条件查询数据时，可以根据分区设定，只读取对应分区文件，达到分区剪枝的效果。
• 列存：如 snapshot 快照数据表，包括上百个字段，但一次聚合查询往往只需要其中几个字段。可以利用 DolphinDB OLAP 引擎的列存技术，在查询时只读取需要的列文件，将极大地减少磁盘 IO 。
• 索引：当分布式表使用 TSDB 引擎时，且查询语句命中 Sort Key时，可以通过扫描稀疏索引文件，来查询对应的数据块ID。进而只读取对应数据块，从而避免全表扫描。

在 DolphinDB 中，可以通过提高批处理数据量的大小、利用多线程并发、使用分布式处理等方式，来提高数据处理速度。

• 合理的批数据量：DolphinDB 以分区为单位管理海量数据，通常数据处理以一个分区为单位，每个分区的大小为100MB-500MB（压缩前）左右比较合适。从 Python 和 MySQL 等技术栈迁移至 DolphinDB 时，需要充分理解分区在 DolphinDB 中的重要性，以便充分发挥 DolphinDB 的性能。推荐阅读分区数据库设计和操作。
• 多线程：现代操作系统及软件使用多线程技术来充分发挥多核服务器的硬件优势。DolphinDB 大量应用了多线程技术，就分布式 SQL 而言，会使用多个线程来并发处理分区数据。
• 分布式：DolphinDB 可以使用多个服务器构建一个水平扩展的分布式集群，并且支持分布式事务。当执行一个分布式 SQL 时，通过 map-merge-reduce 的模型来执行分布式计算。其中 map 阶段会自动调度到集群各个节点，充分利用集群的硬件资源。

按天批量处理所有股票数据，DolphinDB 会将这一天20个分区内的3000个股票的数据，生成20个子任务并行处理，自动调度至集群中的各个节点，进行分布式计算。

优化后代码如下：

def transformData(tradeDate){
	tradeSrc = loadTable("dfs://originData", "trade")
	tradeTgt = loadTable("dfs://formatData", "trade")
	data = select 
		securityID + "SZ" as securityID
		,toIntDate(tradingdate) as  tradingdate
		,tradingtime$STRING as tradingtime
		,recid as recid 
		,tradeprice
		,tradevolume
		,tradevolume * tradeprice as tradeamount     	
		,buyorderid as buyrecid
		,sellorderid as sellrecid
		,iif(sellorderid>  buyorderid,"S", "B") as buysellflag    	
		,unix$LONG as unix
	from tradeSrc
	where tradingdate = tradeDate and tradetype in ["0", "F"]
	tradeTgt.append!(data)
	pnodeRun(flushOLAPCache)
}

allDays = 2022.05.01..2022.05.20
for(aDate in allDays){
	jobId = "transform_"+ strReplace(aDate$STRING, ".", "")	
	jobDesc = "transform data"
	submitJob(jobId, jobDesc, transformData, aDate)
}

以上代码处理一天内3000只股票的数据，大约需要40秒。通过 submitJob 的方式同时处理20个交易日的数据，在 maxBatchJobWorker=16（一般设置为 CPU 核数） 的设置下，总共耗时210秒，性能提升74倍。

相对于原始代码，我们提供的优化代码有以下几点进步：

• 分布式，并行度高：以上 select 部分的代码是分布式、并行执行的；进一步地，可以通过 submitJob 的方式并行执行多个任务。
• 数据访问次数少：所有的处理逻辑，包括过滤数据、类型转换、增加派生字段，通过一次 read 即可完成，无需反复读取数据和对数据进行转换。
• 向量化：OLAP 引擎采用列式存储，一列数据读取到内存中以 vector 形式存储，因此，SQL 运算采用向量化方式，十分高效。