为什么要学数据分析？每个人的初衷许是不同的：有人为了研究工作中的各种数据，有人是苦于团队无人可用只能自己上场，有人是中途转行想要谋求发展，有人则纯粹为了多个傍身技能......尽管目的不同，可学习的过程，都避不开三个字：怎么学。我不是数据分析专家，也不贩卖任何网课，单纯从（走过不少弯路的）同好者的角度，来聊聊这个问题。『怎么学』，要解决的问题包括选择学习渠道、学习规划、学习方法等。首先来谈学习渠道。网课是目前主流的学习渠道，一来相比线下课程而言，费用更低；二来不受空间时间限制，自由度高；三来大神授课可信度高，速成课直接将爽值拉到爆表。五花八门的Excel速成课、价格不菲的python从入门到精通，还有号称7天就能玩转数据的SQL教程、和几分钟甚至几秒的所谓干货短视频，难免看得你眼花缭乱。这么多课程，到底该怎么选？个人建议是，不要急于选课买课，小心交的都是智商税。有比刷课程更重要的事。那就是，明确学习方法、做好学习规划。互联网人在推进业务时，有两个要点：『以终为始』和『目标拆解』。以终为始是一种反向思维方式，就是从最终的结果出发，倒逼策略和规划，同时将最终目标进行拆解，分为多个阶段性目标。关于如何设定目标，个人建议是：要满足『可执行』和『可评估』两个条件。最近在看的《谷歌时间管理课》中，作者也提到：应该避免设定形成目标，而应聚焦在行动目标上。比如，你在学数据分析时，不应把『学会数据分析』作为目标，而应设立类似『15天掌握主要分析方法』、『1个月上手分析数据』、『坚持每周分析一个案例』等更可掌控的行动计划。当然在制定好相应的学习规划之前，需要明确的是学习方法。先问自己一个问题：你学习数据分析，是为了运用数据分析工具，还是运用工具分析数据？发现没有，尽管是一样的文字，调换了顺序，意思却截然不同。你得知道：那些Excel、Python、SQL课程，是工具，而不是方法。你可以刷几分钟短视频，学会在Excel中插入某个函数，可你未必清楚：这一步操作，对最终的数据分析结果、以及数据报告有何意义。本末倒置的学习方法，会带你进入误区，而熟练掌握数据分析框架和分析方法，才能有的放矢，更高效地分析数据、解决问题、优化业务。话不多说，直接呈上相关干货。常用分析框架汇总：QQ模型、用户行为理论、5W2H分析法、AARRR模型、RFM模型、人货场模型。1、QQ模型. (用于分析评估APP某个新增功能的效果）Quantity 数量：用户数
浏览量
点击量Quality 质量：留存率
转化率
参与率案例——分析及评估APP某个新增功能的效果2、用户行为理论5个阶段： 认知 - 熟悉 - 试用 - 使用 - 忠诚案例——蜗牛阅读3、5W2H分析法 （挖掘产品价值，通过研究用户行为，更好地迭代产品）案例——蜗牛阅读4、AARRR模型Refer 推荐 Revenue收入 Retention 用户留存 Activation 用户激活 Acquisition用户获取5、RFM模型R-Recency最近一次购买时间 F-Frequency 购买频次 M-Money 购买金额6、人货场模型常用数据分析方法对比分析法、分组分析法、逻辑树分析法、漏斗图分析法、矩阵关联法。1、对比分析法案例——达达配送（侧重时间对比、空间对比）2、分组分析法案例——3、逻辑树分析法（分层罗列影响因素，发现问题）案例——针对不同层级的点位进行分析，找到决定因素4、漏斗图分析——用于某个行为路径中的问题5、矩阵关联法——对2-4个重要属性进行分析最近也有向相关专家前辈讨教，他们推荐了基本入门级书籍：入门级——《深入浅出数据分析》《深入浅出统计学》《统计数字会撒谎》《用图表说话》《商务与经济统计》《精益数据分析》（初学和老鸟都可以读）技术向——《数据挖掘与数据化运营实战思路、方法》《利用python进行数据分析》《高性能mysql》若有任何学习心得、相关书籍或课程推荐，欢迎分享！最后，附上思维导图，按需自取啦~}

一个量化策略在生产（交易）环境中运行时，处理实时数据的程序通常为事件驱动。为确保研发和生产使用同一套代码，通常在研发阶段需将历史数据，严格按照事件发生的时间顺序进行回放，以此模拟交易环境。一个交易所的行情数据通常包括逐笔委托、逐笔成交、快照等多种数据。DolphinDB 提供了严格按照时间顺序将多个不同数据源同时进行回放的功能。本教程以股票行情回放为例提供一种多表回放方案。首先在第 1、2 章简要介绍 DolphinDB 数据回放功能和原理，之后在第 3 章给出完整的股票行情回放应用方案和代码实现。1. 单表回放DolphinDB 历史数据回放功能通过 replay 函数实现。replay 函数的作用是将内存表或数据库表中的记录以一定的速率写入到目标表中，以模拟实时数据流不断注入目标表的场景。最基础的回放模式是单表回放，即将一个输入表回放至一个相同表结构的目标表中，以下是不包括建表语句的单表回放示例：tradeDS = replayDS(sqlObj=<select * from loadTable("dfs://trade", "trade") where Date = 2020.12.31>, dateColumn=`Date, timeColumn=`Time)
replay(inputTables=tradeDS, outputTables=tradeStream, dateColumn=`Date, timeColumn=`Time, replayRate=10000, absoluteRate=true)以上脚本将数据库 "dfs://trade" 中的 "trade" 表中 2020 年 12 月 31 日的数据以每秒 1 万条的速度注入目标表 tradeStream 中。更多关于 replay、replayDS 函数的介绍可以参考 DolphinDB 历史数据回放教程、replay用户手册、replayDS用户手册。但是，单表回放并不能满足所有的回放要求。因为在实践中，一个领域问题往往需要多个不同类型的消息协作，例如金融领域的行情数据包括逐笔委托、逐笔成交、快照等，为了更好地模拟实际交易中的实时数据流，通常需要将以上三类数据同时进行回放，这时便提出了多表回放的需求。2. 多表回放DolphinDB 在对多表回放的支持上不断地演进，最终提供了基于异构流数据表的异构多表回放方案，其能够实现多个数据源的严格时序回放和消费。本小节将对多表回放面临的难点、相应的 DolphinDB 技术解决方案和原理展开介绍。2.1 多对多回放类似单表回放的原理，replay 函数提供了多对多模式的多表回放，即将多个输入表回放至多个目标表，输入表与目标表一一对应。以下是多对多模式的多表回放的示例：orderDS = replayDS(sqlObj=<select * from loadTable("dfs://order", "order") where Date = 2020.12.31>, dateColumn=`Date, timeColumn=`Time)
tradeDS = replayDS(sqlObj=<select * from loadTable("dfs://trade", "trade") where Date = 2020.12.31>, dateColumn=`Date, timeColumn=`Time)
snapshotDS = replayDS(sqlObj=<select * from loadTable("dfs://snapshot", "snapshot") where Date =2020.12.31>, dateColumn=`Date, timeColumn=`Time)
replay(inputTables=[orderDS, tradeDS, snapshotDS], outputTables=[orderStream, tradeStream, snapshotStream], dateColumn=`Date, timeColumn=`Time, replayRate=10000, absoluteRate=true)以上脚本将三个数据库表中的历史数据分别注入三个目标表中。在多对多的模式中，不同表的在同一秒内的两条数据写入目标表的顺序则可能和数据中的时间字段的先后关系不一致。此外下游如果由三个处理线程分别对三个目标表进行订阅与消费，也很难保证表与表之间的数据被处理的顺序关系。因此，多对多回放不能保证整体上最严格的时序。在实践中，一个领域中不同类型的消息是有先后顺序的，比如股票的逐笔成交和逐笔委托，所以在对多个数据源回放时会有保持每条数据之间的严格的先后顺序的需求，为此我们需要解决以下问题：不同结构的数据如何统一进行排序和注入以保证整体的顺序？如何保证对多表回放结果的实时消费也是严格按照时序进行的？2.2 异构回放面对上述多表回放的难点，DolphinDB 进一步增加了异构模式的多表回放（1.30.17/2.00.5 及以上版本），它将多个不同表结构的数据表写入到同一张异构流数据表中，从而实现了严格按时间顺序的多表回放。以下是异构模式的多表回放示例：orderDS = replayDS(sqlObj=<select * from loadTable("dfs://order", "order") where Date = 2020.12.31>, dateColumn=`Date, timeColumn=`Time)
tradeDS = replayDS(sqlObj=<select * from loadTable("dfs://trade", "trade") where Date = 2020.12.31>, dateColumn=`Date, timeColumn=`Time)
snapshotDS = replayDS(sqlObj=<select * from loadTable("dfs://snapshot", "snapshot") where Date =2020.12.31>, dateColumn=`Date, timeColumn=`Time)
inputDict = dict(["order", "trade", "snapshot"], [orderDS, tradeDS, snapshotDS])
replay(inputTables=inputDict, outputTables=messageStream, dateColumn=`Date, timeColumn=`Time, replayRate=10000, absoluteRate=true)异构回放时将 replay 函数的 inputTables 参数指定为字典，outputTables 参数指定为异构流数据表。inputTables 参数指定多个结构不同的数据源，字典的 key 是用户自定义的字符串，是数据源的唯一标识，将会对应 outputTables 参数指定的表的第二列，字典的 value 是通过 replayDS 定义的数据源或者表。以上脚本中的输出表 messageStream 为异构流数据表，其表结构如下：nametypeStringcommentmsgTimeTIMESTAMP消息时间msgTypeSYMBOL消息类型："order"、"trade"、"snapshot"msgBodyBLOB消息内容，以二进制格式存储异构回放时 outputTables 参数指定的表至少需要包含以上三列。回放完成后，表 messageStream 的数据预览如下：表中每行记录对应输入表中的一行记录，msgTime 字段是输入表中的时间列，msgType 字段用来区分来自哪张输入表，msgBody 字段以二进制格式存储了输入表中的记录内容。异构回放基于异构流数据表对多个数据源全局排序并写入目标表，因而保证了多个数据源之间的严格时间顺序。同时，异构流数据表和普通流数据表一样可以被订阅，即多种类型的数据存储在同一张表中被发布并被同一个线程实时处理，因而也保证了消费的严格时序性。为了对异构流数据表进行指标计算等数据处理，则需要将二进制格式的消息内容反序列化为原始结构的一行记录。DolphinDB 在脚本语言以及在 API 中均支持了对异构流数据表的解析功能。在脚本中，使用流过滤与分发引擎 streamFilter 对异构流数据表进行反序列化，并对反序列后的结果指定数据处理逻辑；在 API 中，各类 API 在支持流数据订阅功能的基础上，扩展支持了在订阅时对异构流数据表进行反序列化。3. 多表回放应用：股票行情回放基于上文提到的异构回放、异构流数据表解析以及 DolphinDB 流处理框架中的其他特性等，本章将结合股票行情回放展示 DolphinDB 异构模式的多表回放功能在实际场景中的应用，包括数据回放以及三种具体的回放结果消费方案。3.1 行情回放与消费方案行情多表回放方案的数据处理流程图如下：处理流程图说明：回放与消费流程围绕异构流数据表 messageStream 展开。图中异构流数据表模块以上，为异构模式的多表回放的示意图，由数据回放工具即 replay 和 replayDS 函数，将存储在多个数据库中的原始数据回放至异构流数据表中。图中异构流数据表模块以下，分为三条支路，分别对应对回放结果的三种不同的处理方式，从左至右依次是：在 DolphinDB 的订阅中，通过内置的流计算引擎实时计算指标，本文将使用 asof join 引擎实时关联逐笔成交与快照数据，计算个股交易成本并写入结果流数据表；在 DolphinDB 的订阅中，通过 Kafka 插件将回放结果实时写入外部的消息中间件 Kafka；在外部程序中，通过 DolphinDB 的流数据 API 来实时订阅和消费回放结果，本文将使用 C++API。3.2 测试数据集本教程基于上交所某日的股票行情数据进行回放，包括逐笔委托、逐笔成交、Level2 快照三种行情数据，分别存放在分布式数据库 "dfs://order"、"dfs://trade"、"dfs://snapshot" 中，均使用 TSDB 存储引擎，数据概览如下：数据集字段数总行数数据大小简称分区机制排序列逐笔委托20490185526.8GorderVALUE: 交易日, HASH: [SYMBOL, 20]股票, 交易时间逐笔成交15436527183.3GtradeVALUE: 交易日, HASH: [SYMBOL, 20]股票, 交易时间Level2 行情快照5584103594.1GsnapshotVALUE: 交易日, HASH: [SYMBOL, 20]股票, 交易时间后文也将提供部分原始数据的 csv 文件以及对应的数据导入脚本，以便快速体验回放功能。3.3 代码实现本教程代码开发工具采用 DolphinDB GUI，所有代码均可在 DolphinDB GUI 客户端开发工具执行。相关环境配置见后文 [5. 开发环境配置](#5 - 开发环境配置)。3.3.1 股票行情回放此部分代码为 DolphinDB 脚本，将三个数据库中的不同结构的数据回放至同一个异构流数据表中。完整脚本见附录 01. 股票行情回放. txt。创建异构流数据表 messageStreamcolName = `msgTime`msgType`msgBody
colType = [TIMESTAMP,SYMBOL, BLOB] messageTemp = streamTable(1000000:0, colName, colType)
enableTableShareAndPersistence(table=messageTemp, tableName="messageStream", asynWrite=true, compress=true, cacheSize=1000000, retentionMinutes=1440, flushMode=0, preCache=10000)
messageTemp = NULLmessageStream 是共享的异步持久化异构流数据表。共享意味着在当前节点的所有会话中可见，为了之后能够对该表进行订阅，必须将其定义为共享的流数据表。同时此处对流数据表进行持久化，其主要目的是控制该表的最大内存占用，enableTableShareAndPersistence 函数中的 cacheSize 参数规定了该表在内存中最多保留 100 万行。流数据持久化也保障了流数据的备份和恢复，当节点异常关闭后，持久化的数据会在重启时自动载入流数据表以继续流数据消费。三个数据源异构回放至流数据表 messageStreamtimeRS = cutPoints(09:15:00.000..15:00:00.000, 100)
orderDS = replayDS(sqlObj=<select * from loadTable("dfs://order", "order") where Date = 2020.12.31>, dateColumn=`Date, timeColumn=`Time, timeRepartitionSchema=timeRS)
tradeDS = replayDS(sqlObj=<select * from loadTable("dfs://trade", "trade") where Date = 2020.12.31>, dateColumn=`Date, timeColumn=`Time, timeRepartitionSchema=timeRS)
snapshotDS = replayDS(sqlObj=<select * from loadTable("dfs://snapshot", "snapshot") where Date =2020.12.31>, dateColumn=`Date, timeColumn=`Time, timeRepartitionSchema=timeRS)
inputDict = dict(["order", "trade", "snapshot"], [orderDS, tradeDS, snapshotDS])
submitJob("replay", "replay stock market", replay, inputDict, messageStream, `Date, `Time, , , 3)上述脚本读取三个数据库中的结构不同的数据进行全速的异构回放，回放通过 submitJob 函数提交后台作业来执行。下面讲解对于回放进行调优的相关参数和原理：replayDS 函数中的 timeRepartitionSchema 参数：replayDS 函数将输入的 SQL 查询转化为数据源，其会根据输入表的分区以及 timeRepartitionSchema 参数，将原始的 SQL 查询按照时间顺序拆分成若干小的 SQL 查询。replay 函数中的 parallelLevel 参数：parallelLevel 表示从数据库查询数据的并行度，即同时查询经过划分之后的小数据源的线程数，默认为 1，上述脚本中通过 submitjob 的参数设置为 3。对数据库表的回放过程分为两步，其一是通过 SQL 查询历史数据至内存，查询包括对数据的排序，其二是将内存中的数据写入输出表，两步以流水线的方式执行。若将某日数据全部导入内存并排序，会占用大量内存甚至导致内存不足，同时由于全部数据的查询耗时比较长，会导致第二步写入输出表的时刻也相应推迟。以 orderDS 为例，若不设置 timeRepartitionSchema 参数，则相应的 SQL 查询为：select * from loadTable("dfs://order", "order") where Date = 2020.12.31 order by Time因此针对大数据量的场景，本教程先对 replayDS 函数指定 timeRepartitionSchema 参数，将数据按照时间戳分为 100 个小数据源，则每次查询的最小单位为其中一个小数据源，同时提高 parallelLevel 参数来帮助提升查询速度。以 orderDS 为例，若设置上述 timeRepartitionSchema 参数，则相应的其中一个 SQL 查询为：select * from loadTable("dfs://order", "order") where Date = 2020.12.31, 09:15:00.000 <= Time < 09:18:27.001 order by Time后文的 [4. 性能测试](#4 - 性能测试) 章节利用本节的脚本进行了性能测试，最终总耗时 4m18s，内存占用峰值 4.7GB。内存主要由回放过程中的 SQL 查询和输出表等占用，通过对数据源进行切割以及对输出表进行持久化能有效控制内存使用。作业运维：在 submitJob 函数提交后，通过 getRecentJobs 函数可以查看后台作业的状态，如果 endTime 和 errorMsg 为空，说明任务正在正常运行中。也可以用 cancelJob 函数取消回放，其输入参数 jobId 通过 getRecentJobs 获取。若没有可作为数据源的数据库，也可以通过加载 csv 文件至内存中进行回放来快速体验本教程，附录中的数据文件提供了 100 支股票的某日完整行情数据，全部数据在内存中约占 700M。以下脚本需要修改 loadText 的路径为实际的 csv 文本数据存储路径。orderDS = select * from loadText("/yourDataPath/replayData/order.csv") order by Time
tradeDS = select * from loadText("/yourDataPath/replayData/trade.csv") order by Time
snapshotDS = select * from loadText("/yourDataPath/replayData/snapshot.csv") order by Time
inputDict = dict(["order", "trade", "snapshot"], [orderDS, tradeDS, snapshotDS])
submitJob("replay", "replay text", replay, inputDict, messageStream, `Date, `Time, , , 1)3.3.2 消费场景 1：在 DolphinDB 订阅中实时计算个股交易成本此部分代码为 DolphinDB 脚本，将实时消费 3.3.1 小节创建的流数据表 messageStream，使用 asof join 引擎实时关联逐笔成交与快照数据，并计算个股交易成本，完整脚本见附录 02. 消费场景 1: 计算个股交易成本_asofJoin 实现. txt。创建计算结果输出表 prevailingQuotescolName = `TradeTime`SecurityID`Price`TradeQty`BidPX1`OfferPX1`Spread`SnapshotTime
colType = [TIME, SYMBOL, DOUBLE, INT, DOUBLE, DOUBLE, DOUBLE, TIME]
prevailingQuotesTemp = streamTable(1000000:0, colName, colType)
enableTableShareAndPersistence(table=prevailingQuotesTemp, tableName="prevailingQuotes", asynWrite=true, compress=true, cacheSize=1000000, retentionMinutes=1440, flushMode=0, preCache=10000)
prevailingQuotesTemp = NULLprevailingQuotes 被定义为共享流数据表，之后可以对其进行订阅和进一步的处理。创建流计算 asof join 引擎def createSchemaTable(dbName, tableName){
schema = loadTable(dbName, tableName).schema().colDefs
return table(1:0, schema.name, schema.typeString)
}
tradeSchema = createSchemaTable("dfs://trade", "trade")
snapshotSchema = createSchemaTable("dfs://snapshot", "snapshot")
joinEngine=createAsofJoinEngine(name="tradeJoinSnapshot", leftTable=tradeSchema, rightTable=snapshotSchema, outputTable=prevailingQuotes, metrics=<[Price, TradeQty, BidPX1, OfferPX1, abs(Price-(BidPX1+OfferPX1)/2), snapshotSchema.Time]>, matchingColumn=`SecurityID, timeColumn=`Time, useSystemTime=false, delayedTime=1)使用 asof join 引擎实现在对股票分组的基础上，对于每条输入的 trade 记录，实时关联与之在时间列上最接近的一条 snapshot 记录，并使用 trade 中的价格字段和 snapshot 中的报价字段进行指标计算。最终，以上配置的 asof join 引擎会输出和左表行数相同的结果。asof join 引擎更多介绍请参考 createAsofJoinEngine 用户手册。值得一提的是，asof join 引擎用于两个数据流的实时关联，其优势在于支持通过事件时间进行关联，即这里的配置项 useSystemTime=false 表示关联时使用数据中的时间列，在实际应用中，考虑到在两个数据流中相同事件时间的数据的到来可能发生乱序，使用事件时间进行关联通常是更符合业务含义的做法。在此处，异构回放能够保证两个数据流的严格处理顺序，因此也可以使用 look up join 引擎得到相同的计算结果，即用 trade 表去关联在系统时间上最新的一条 snapshot 记录，此处使用 look up join 引擎会比 asof join 引擎在性能上有稍好的表现，脚本见附录 03. 消费场景 1: 计算个股交易成本_lookUpJoin 实现. txt。look up join 引擎更多介绍请参考 createLookUpJoinEngine 用户手册。自定义函数 createSchemaTable，用于获取数据库表的表结构，以做为创建引擎时的参数传入。创建流计算过滤与分发引擎def appendLeftStream(msg){
tempMsg = select * from msg where Price > 0 and Time>=09:30:00.000
getLeftStream(getStreamEngine(`tradeJoinSnapshot)).tableInsert(tempMsg)
}
filter1 = dict(STRING,ANY)
filter1["condition"] = "trade"
filter1["handler"] = appendLeftStream
filter2 = dict(STRING,ANY)
filter2["condition"] = "snapshot"
filter2["handler"] = getRightStream(getStreamEngine(`tradeJoinSnapshot))
schema = dict(["trade", "snapshot"], [tradeSchema, snapshotSchema])
parseEngine = streamFilter(name="streamFilter", dummyTable=messageStream, filter=[filter1, filter2], msgSchema=schema)streamFilter 函数通过设置 msgSchema 参数，会对异构流数据表进行反序列，并根据 filter 参数中设置的 handler 来处理订阅的数据。当订阅数据到来时，handler 之间是串行执行的，保证了对数据的处理严格按照时序进行。handler 参数是一元函数或数据表，用于处理订阅的数据。当它是函数时，其唯一的参数是经过解析和过滤后的数据表。这里对于 trade 数据其 handler 设置为函数 appendLeftStream，该函数对订阅到的数据首先做过滤，再将符合条件的数据作为左表写入到 asof join 引擎。对于 snapshot 数据其 handler 设置为 asof join 引擎的右表，表示对于订阅到的 snapshot 数据直接作为右表写入 asof join 引擎。订阅异构流数据表subscribeTable(tableName="messageStream", actionName="tradeJoinSnapshot", offset=-1, handler=engine, msgAsTable=true, reconnect=true)设置参数 offset 为 - 1，订阅将会从提交订阅时流数据表的当前行开始。为了消费到完整的数据，建议先执行此处脚本以提交订阅，再提交 3.3.1 小节的后台回放作业。在 GUI 中查看计算结果asof join 引擎在 matchingColumn 的分组内输出表数据与输入时的顺序一致。3.3.3 消费场景 2：在 DolphinDB 订阅中将回放结果实时推送 Kafka此部分代码为 DolphinDB 脚本，将 3.3.1 小节创建的流数据表 messageStream 实时发送至消息中间件 Kafka。执行以下脚本，需要有可以写入的 Kafka server，并且安装 DolphinDB Kafka 插件。完整脚本见附录 04. 消费场景 2: 实时推送 Kafka.txt。加载 Kafka 插件并创建 Kafka producerloadPlugin("/DolphinDB/server/plugins/kafka/PluginKafka.txt")
go
producerCfg = dict(STRING, ANY)
producerCfg["metadata.broker.list"] = "localhost"
producer = kafka::producer(producerCfg)loadPlugin 的路径和 producer 配置请按需修改。本教程涉及的 Kafka server 和 DolphinDB server 在同一台服务器上，故 metadata.broker.list 参数为 localhost。定义推送数据至 Kafka topic 的函数def sendMsgToKafkaFunc(dataType, producer, msg){
startTime = now()
try {
kafka::produce(producer, "topic-message", 1, msg, true)
cost = now() - startTime
writeLog("[Kafka Plugin] Successed to send" + dataType + ":" + msg.size() + "rows," + cost + "ms.")
}
catch(ex) {writeLog("[Kafka Plugin] Failed to send msg to kafka with error:" +ex)}
}kafka::produce 函数会将任意表结构的 msg 以 json 格式发送至指定的 Kafka topic。此处的 writeLog 函数在日志中打印每批推送的情况来方便运维观察。注册流数据过滤与分发引擎filter1 = dict(STRING,ANY)
filter1["condition"] =
"order"
filter1["handler"] = sendMsgToKafkaFunc{"order", producer}
filter2 = dict(STRING,ANY)
filter2["condition"] = "trade"
filter2["handler"] = sendMsgToKafkaFunc{"trade", producer}
filter3 = dict(STRING,ANY)
filter3["condition"] = "snapshot"
filter3["handler"] = sendMsgToKafkaFunc{"snapshot", producer}
schema = dict(["order","trade", "snapshot"], [loadTable("dfs://order", "order"), loadTable("dfs://trade", "trade"), loadTable("dfs://snapshot", "snapshot")])
filterEngine = streamFilter(name="streamFilter", dummyTable=messageStream, filter=[filter1, filter2, filter3], msgSchema=schema)streamFilter 函数通过设置 msgSchema 参数，会对异构流数据表进行反序列，并根据 filter 参数中设置的 handler 来处理订阅的数据。当订阅数据到来时，handler 之间是串行执行的，保证了对数据的处理严格按照时序进行。handler 参数是一元函数或数据表，用于处理订阅的数据。当它是函数时，其唯一的参数是经过解析和过滤后的数据表。sendMsgToKafka{"order", producer} 的写法是函数化编程中的部分应用，即指定函数 sendMsgToKafka 的部分参数，产生一个参数较少的新函数。订阅异构流数据表subscribeTable(tableName="messageStream", actionName="sendMsgToKafka", offset=-1, handler=engine, msgAsTable=true, reconnect=true)设置参数 offset 为 - 1，订阅将会从提交订阅时流数据表的当前行开始。为了消费到完整的数据，建议先执行此处脚本以提交订阅，再提交 3.3.1 小节的后台回放作业。在终端查看发送结果在命令行开启消费者进程，从第一条开始消费名为 topic-message 的 topic./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic topic-message返回：3.3.4 消费场景 3：在外部程序中通过 C++API 实时订阅与处理此部分代码为 C++ 程序，程序会订阅 3.3.1 小节创建的异构流数据表 messageStream，并实时打印每条数据。完整代码见附录 05. 消费场景 3:C++API 实时订阅. cpp。int main(int argc, char *argv[]){
DBConnection conn;
string hostName = "127.0.0.1";
int port = 8848;
bool ret = conn.connect(hostName, port);
conn.run("login(\"admin\", \"123456\")");
DictionarySP t1schema = conn.run("loadTable(\"dfs://snapshotL2\", \"snapshotL2\").schema()");
DictionarySP t2schema = conn.run("loadTable(\"dfs://trade\", \"trade\").schema()");
DictionarySP t3schema = conn.run("loadTable(\"dfs://order\", \"order\").schema()");
unordered_map<string, DictionarySP> sym2schema;
sym2schema["snapshot"] = t1schema;
sym2schema["trade"] = t2schema;
sym2schema["order"] = t3schema;
StreamDeserializerSP sdsp = new StreamDeserializer(sym2schema);
auto myHandler = [&](Message msg) {
const string &symbol = msg.getSymbol();
cout << symbol << ":";
size_t len = msg->size();
for (int i = 0; i < len; i++) {
cout <<msg->get(i)->getString() << ",";
}
cout << endl;
};
int listenport = 10260;
ThreadedClient threadedClient(listenport);
auto thread = threadedClient.subscribe(hostName, port, myHandler, "messageStream", "printMessageStream", -1, true, nullptr, false, false, sdsp);
cout<<"Successed to subscribe messageStream"<<endl;
thread->join();
return 0;
}调用订阅函数 ThreadedClient::subscribe 订阅异构流数据表时，在最后一个参数指定相应的流数据反序列化实例 StreamDeserializerSP，则在订阅时会对收到的数据进行反序列化再传递给用户自定义的回调函数 myHandler。listenport 参数为单线程客户端的订阅端口号，设置 C++ 程序所在服务器的任意空闲端口即可。在终端查看程序实时打印的内容：3.3.5 清理环境如果反复执行上述脚本，可能需要清理流数据表、取消订阅、注销流计算引擎等操作，建议在运行回放与消费脚本前先清理环境。本教程的清理环境脚本见附录 06. 清理环境. txt。4. 性能测试本教程对异构模式下的多表回放功能进行了性能测试。以 3.2 小节的测试数据集作为回放的输入，以 3.3.1 小节的回放脚本作为测试脚本，该脚本不设定回放速率（即以最快的速率回放），并且输出表没有任何订阅，最终回放了 101,081,629 条数据至输出表中，总耗时 4m18s，每秒回放约 39 万条数据，内存占用峰值 4.7GB。测试使用的服务器的 CPU 为 Intel(R) Xeon(R) Silver 4216 CPU @ 2.10GHz。5. 开发环境配置服务器环境CPU 类型：Intel(R) Xeon(R) Silver 4216 CPU @ 2.10GHz逻辑 CPU 总数：8内存：64GBOS：64 位 CentOS Linux 7 (Core)DolphinDB server 部署server 版本：2.00.6server 部署模式：单节点配置文件：dolphindb.cfglocalSite=localhost:8848:local8848
mode=single
maxMemSize=32
maxConnections=512
workerNum=8
localExecutors=7
maxConnectionPerSite=15
newValuePartitionPolicy=add
webWorkerNum=2
dataSync=1
persistenceDir=/DolphinDB/server/persistenceDir
maxPubConnections=64
subExecutors=16
subPort=8849
subThrottle=1
persistenceWorkerNum=1
lanCluster=0配置参数 persistenceDir 需要开发人员根据实际环境配置。单节点部署教程：单节点部署DolphinDB client 开发环境CPU 类型：Intel(R) Core(TM) i7-7700 CPU @ 3.60GHz 3.60 GHz逻辑 CPU 总数：8内存：32GBOS：Windows 10 专业版DolphinDB GUI 版本：1.30.15DolphinDB GUI 安装教程：GUI 教程DolphinDB Kafka 插件安装Kafka 插件版本：release200Kafka 插件版本建议按 DolphinDB server 版本选择，如 2.00.6 版本的 server 安装 release200 分支的插件Kafka 插件教程：Kafka 插件教程Kafka server 部署zookeeper 版本：3.4.6Kafka 版本：2.12-2.6.2Kafka 部署模式：单节点创建 Kafka topic./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 4 --topic topic-messageDolphinDB C++ API 安装C++ API 版本：release200C++API 版本建议按 DolphinDB server 版本选择，如 2.00.6 版本的 server 安装 release200 分支的 APIC++ API 教程：C++ API 教程6. 总结本教程基于数据回放工具、流数据处理、异构流数据表等 DolphinDB 特性，提供了多表股票行情数据回放以及回放结果落地方案，意在使开发人员可以在此基础上快速搭建自己的行情回放系统。附录01.股票行情回放.txt02.消费场景1：计算个股交易成本 _asofJoin 实现.txt03.消费场景1：计算个股交易成本 _lookUpJoin 实现.txt04.消费场景2：实时推送 Kafka.txt05.消费场景3：C++API 实时订阅.cpp06.清理环境.txt样例数据：order.csv样例数据：snapshot.csv样例数据：trade.csv}

由于数据应用开发和功能性软件系统开发存在很大的不同，在我们实践过程中，在开发人员和质量保证人员间常常有大量关于测试如何实施的讨论。下文将尝试总结一下数据应用开发的特点，并讨论在这些特点之下，对应的测试策略应该是怎么样的。功能性软件的测试先来回顾一下功能性软件系统开发中的测试。测试一般分为自动化测试和手工测试。由于手工测试对人工依赖程度很高，如果主要依赖手工测试来保证软件质量，将无法满足软件快速迭代上线的需要。现代软件开发越来越强调自动化测试的作用，这也是敏捷软件开发的基本要求。有了全方位的自动化测试保障，就有可能做到每周上线，每日上线甚至随时上线。这里主要讨论自动化测试。测试金字塔我们一般会按照如下测试金字塔的原则来组织自动化测试。测试金字塔分为三层，自下而上分别对应单元测试、集成测试、端到端测试。单元测试是指函数或类级别的，较小范围代码的测试，一般不依赖外部系统（可通过Mock或测试替身等实现）。单元测试的特点是运行速度非常快（最好全部在内存中运行），所以执行这种测试的成本也就很低。单元测试在测试金字塔的最底端，占的面积最大。这指导我们应该构建大量的这类测试，并以这类测试为主来保证软件质量。集成测试是比单元测试集成程度更高的测试，它在运行时执行的代码路径更广，通常会依赖数据库、文件系统等外部环境。由于依赖了外部环境，集成测试的运行速度更慢，执行测试的成本更高。集成测试在测试金字塔的中间，这指导我们应该构建中等数量的这类测试。集成测试在Web应用场景中也常常被称为服务测试（Service Test）或API测试。端到端测试是比集成测试更靠后的测试，通常通过直接模拟用户操作来构建这样的测试。由于需要模拟用户操作，所以它常常需要依赖一整套完整集成好的环境，这样一来，其运行速度也是最慢的。端到端测试在Web应用场景中也常常被称为UI测试。端到端测试在测试金字塔的顶端，这指导我们应该构建少量的这类测试。测试的范围非常广，实施方法也非常灵活。哪里是重点？我们要在哪里发力？测试金字塔为我们指明了方向。进入测试金字塔为了更深入地理解一般软件的测试要怎么做，我们需要进一步深入分析一下测试金字塔。测试带来的信心上文中的金字塔图示有一个特点并没有反映出来，那就是，越上层的测试给团队带来的信心越强。这还算好理解，试想，如果没有单元测试，只有端到端测试，我们是不是可以认为程序大部分还是可以正常工作的（可能存在一些边界场景有问题）？但是如果只有单元测试而没有端到端测试，我们连程序能不能运行都不知道！端到端测试能带来很强的信心，但这常常构成另一个陷阱。由于端到端测试对团队有很大的吸引力，一些团队可能会选择直接构建大量的端到端测试而忽略单元测试。这些端到端测试运行缓慢，一般也难以修改，很快就会让团队举步维艰。缓慢的测试带来了缓慢的持续集成，高频率的上线就慢慢变得遥不可及。单元测试虽然不能直接给人很强的信心，但是常常是更有效的测试手段，因为它可以很容易的覆盖到各种边界场景。测试金字塔是敏捷软件开发所推崇的测试原则，它是在测试带来的信心和测试本身的可维护性两者中权衡做出的选择。测试金字塔可以指导我们构建足够的测试，使得团队既对软件质量有足够的信心，又不会有太多的测试维护负担。既然是权衡，那么我们是否可以以单元测试和集成测试为主，而根本不构建端到端测试（此时端到端测试的功能通过手工测试完成）呢？测试集成度对于一些没有UI（或者说GUI）的应用，或者一些程序库、框架（如Spring）等，很多时候测试金字塔中的三类测试并不直接适用。我们可以这样理解：测试金字塔并非只是三层，它更多的是帮我们建立了在项目中组织测试的原则。事实上，对于通用的软件测试，我们可以理解为存在一个集成度的属性。沿着金字塔往上，测试的集成度越高（依赖外部组件越多）。由于集成度更高，测试过程所要运行的代码就更多更复杂，测试运行时间就越长，测试构建和维护成本就越高。实践过程中，为了提高软件质量和可维护性，我们应当构建更多集成度低的测试。有了测试集成度的理解，我们就可以知道，其实金字塔可以不是三层，它完全可以是两层或者四层、五层。这取决于我们怎么划定某一类测试的范围。同时，我们还可以知道，其实单元测试、集成测试与端到端测试其实并没有特别明显的界限。下面，我们从测试集成度的角度来看如何构建单元测试。上文提到，测试最好通过Mock或测试替身等实现，从而可以不依赖外部系统。但是，如果测试Mock或测试替身难以构造，或者构造之后我们发现测试代码和产品代码耦合非常严重，这时应该怎么办呢？一个可能的选择是考虑使用更高集成度的测试。Spark程序就是这样的一个例子。一旦使用了Spark的DataFrame API去编写代码，我们就几乎无法通过Mock Spark的API或构造一个Spark测试替身的方式编写测试。这时的测试就只能退一步选择集成度更高一些的测试，比如，启动一个本地的Spark环境，然后在这个环境中运行测试。此时，上面的测试属于哪种测试呢？如果我们用三层测试金字塔的测试划分来看待问题，就很难给这样的测试一个准确的定位。不过，通常我们无需考虑这样的分类，而是可以把它当做集成度低的测试，即金字塔靠底端的测试。如果团队成员能达成一致，我们可以称其为单元测试，如果不能，称其为Spark测试也并非不可。何时停止测试所以，对于一般的软件测试，我们可以认为测试策略应当符合一般意义的金字塔。金字塔的细节，比如应该有几层塔，每一层的范围应该是什么样，每一层应该用什么样的测试技术等等，这些问题需要根据具体的情况进行抉择。在讨论一般软件的测试时，需要关注软件的测试何时停止，即，如何判断软件测试已经足够了呢？在老马的《重构 第二版》中，有对于何时停止测试的观点：有一些测试规则建议会尝试保证我们测试一切的组合，虽然这些建议值得了解，但是实践中我们需要适可而止，因为测试达到一定程度之后，其边际效用会递减。如果编写太多测试，我们可能因为工作量太大而气馁。我们应该把注意力集中在最容易出错的地方，最没有信心的地方。一些测试的指标，如覆盖率，能一定程度上衡量测试是否全面而有效，但是最佳的衡量方式可能来自于主观的感受，如果我们觉得对代码比较有信心，那就说明我们的测试做的不错了。主观的信心指数可能是衡量测试是否足够的重要参考。如果要问测试是否足够，我们要自问是否有信心软件能正常工作。在实践过程中，我们还可以尝试分析每次bug出现的原因，如果是由于大部分bug是由于代码没有测试覆盖而产生的，此时我们可能应该编写更多的测试。但如果是由于其他的原因，比如需求分析不足或场景设计不完备而导致的，则应该在对应的阶段做加强，而不是一味的去添加测试。数据应用的测试有了前面对测试策略的分析，我们来看看数据应用的测试策略。数据应用相比功能性软件有很大的不同，但数据应用也属于一般意义上的软件。数据应用有哪些特点，应该如何针对性的做测试呢？下面我们来探讨一下这几个问题。根据前面的文章分析，数据应用中的代码可以大致分为四类：基础框架（如增强SQL执行器）、以SQL为主的ETL脚本、SQL自定义函数（udf）、数据工具（如前文提到的DWD建模工具）。基础框架的测试基础框架代码是数据应用的核心代码，它不仅逻辑较为复杂，而且需要在生产运行时支持大量的ETL的运行。谁也不想提交了有问题的基础框架代码而导致大规模的ETL运行失败。所以我们应当非常重视基础框架的测试，以保证这部分代码的高质量。基础框架的代码通常由Python或Scala编写，由于Python和Scala语言本身都有很好的测试支持，这十分有利于我们做测试。基础框架的另一个特点是它通常没有GUI。按照测试金字塔原理，我们应当为其建立更多的集成度低的测试（下文称单元测试）以及少量的集成度高的测试（下文称集成测试）。比如，在前面的文章中，我们增强了SQL的语法，加入了变量、函数、模板等新的语法元素。在运行时进行变量替换、函数调用等等功能通过基础框架实现。这部分功能逻辑较为复杂，应当建立更多的单元测试及少量的集成测试。ETL脚本的测试ETL脚本的测试可能是数据应用中的最大难点。采用偏集成的测试ETL脚本一般基于SQL实现。SQL本身是一个高度定制化的DSL，如同XML配置一样。XML要如何测试？很多团队可能会直接忽略这类测试。但是用SQL编写的ETL代码有时候还是可以达到几百行的规模，有较多的逻辑，不测试的话难以给人以信心。如何测试呢？如果采用基于Mock的方法写测试，我们会发现测试代码跟产品代码是一样的。所以，这样做意义不大。如果采用高集成度的测试方式（下文称集成测试），即运行ETL并比对结果，我们将发现测试的编写和维护成本都较高。由于ETL脚本代码本身可能是比较简单且不易出错的，为了不易出错的代码编写测试本身就必要性不高，更何况测试的编写和维护成本还比较高。这就显得集成测试这一做法事倍功半。这里可以举一个例子。比如对于一个分组求和并排序输出的SQL，它的代码可能是下图这样的。如果我们去准备输入数据和输出数据，考虑到各种数据的组合场景，我们可能会花费很多的时间，这带来了较高的测试编写成本。并且，当我们要修改SQL时，我们还不得不修改测试，这带来了维护成本。当我们要运行这个测试时，我们不得不完成建表、写数据、运行脚本、比对结果的整个过程。这些过程都需要依赖外部系统，从而导致测试运行缓慢。这也是高维护成本的体现。可见这两种测试方式都不是好的测试方式。测试构建原则那么有没有什么好的原则呢？我们从实践中总结出了几点比较有价值的思路供大家参考。将ETL脚本分为简单ETL和复杂ETL（可以通过代码行数，数据筛选条件多少等进行衡量）。简单的ETL通过代码评审或结对编程来保证代码质量，不做自动化测试。复杂的ETL通过建立集成测试来保证质量。由于集成测试运行较慢，可以考虑：
尽量少点用例数量，将多个用例合并为一个来运行（主要是将数据可以合并成单一的一套数据来运行）将测试分级为需要频繁运行的测试和无需频繁运行的测试，比如可将测试分级P0-P5，P3-P5是经常（如每天或每次代码提交）要运行的测试，P0-P2可以低频（如每周）运行开发测试支持工具，使得运行时可以尽量脱离缓慢的集群环境。如使用Spark读写本地表考虑将复杂的逻辑使用自定义函数实现，降低ETL脚本的复杂度。对自定义函数建立完整的单元测试。将复杂的ETL脚本拆分为多个简单的ETL脚本实现，从而降低单个ETL脚本的复杂度。加深对业务和数据的理解我们在实践过程中发现，其实大多数时候ETL脚本的问题不在于代码写错了，而在于对业务和数据理解不够。比如，前面文章中的空调销售的例子，如果我们在统计销量的时候不知道存在退货或者他店调货的业务实际情况，那我们就不知道数据中还有一些字段能反映这个业务，也就不能正确的计算销量了。想要形成对数据的深入理解需要对长时间的业务知识积累和长时间对数据的探索分析（业务系统通常经历了长时间的发展，在此期间内业务规则复杂性不断增加，导致数据的复杂性不断增加）。对于刚加入团队的新人，他们更容易由于没有考虑到某些业务情况而导致数据计算错误。加深对业务和数据的理解是进行高效和高质量ETL脚本开发的必由之路。有没有什么好的实践方法可以帮助我们加深理解呢？以下几点是我们在实践中总结的值得参考的建议：通过思维导图/流程图来整理复杂的业务流程（或业务知识），形成知识库尽量多的进行数据探索，发掘容易忽略的领域业务知识，并通过第一步进行记录找业务系统团队沟通，找出更多的领域业务知识，并通过第一步进行记录如果有条件，可以更频繁的实地使用业务系统，总结更多的领域业务知识，并通过第一步进行记录针对第一步搜集到的这些容易忽略的特定领域业务流程，设计自动化测试用例进行覆盖SQL自定义函数的测试在基于Hadoop的分布式数据平台环境下，SQL自定义函数通常通过Python或Scala编写。由于这些代码通常对外部的依赖很少，通常只是单纯的根据输入数据计算得到输出数据，所以对这些代码建立测试是十分容易的事。事实上，我们很容易实现100%的测试覆盖率。在组织测试时，我们可以用单元测试的方式，不依赖计算框架。比如，以下Scala编写的自定义函数：对其建立测试时，可以直接测试内部的转换函数array_join_f，一些示例的测试场景比如：在建立了单元测试之后，一般还需要考虑建立少量的集成测试，即通过Spark框架运行SQL来测试此自定义函数，一个示例可以是：如果自定义函数本身十分简单，我们也可以直接通过Spark测试来覆盖所有场景。从上面的讨论可以看出，SQL自定义函数是很容易测试的。除了好测试之外，SQL自定义函数还有很多好的特性，比如可以很好的降低ETL复杂度，可以很方便的被复用等。所以，我们应该尽量考虑将复杂的业务逻辑通过自定义函数封装起来。这也是业界数据开发所建议的做法（大多数的数据开发框架都对自定义函数提供了很好的支持，如Hive Presto ClickHouse等，大多数ETL开发工具也都支持自定义函数的开发）。数据工具的测试数据工具的实例可以参考文章《数据仓库建模自动化》和《数据开发支持工具》。这些工具的一大特点是，它们是用于支持ETL开发的，仅在开发过程中使用。由于它们并不是在产品环境中运行的代码，所以我们可以降低对其的质量要求。这些工具通常只是开发人员为了提高开发效率而编写的代码，存在较大的修改和重构的可能，所以，过早的去建立较完善的测试必要性不高。在我们的实践过程中，这类代码通常只有很少的测试，我们只对那些特别复杂、没有信心能正确工作的地方建立单元测试。如果这些工具代码是通过TDD的方式编写的，通常其测试会更多一些。在持续集成流水线中运行测试前面我们讨论了如何针对数据应用编写测试，还有一个关于测试的重要话题，那就是如何在持续交付流水线中运行这些测试。在功能性软件项目中，如果我们按照测试金字塔的三层来组织测试，那么在流水线中一般就会对应三个测试过程。从上面的讨论可知，数据应用的测试被纵向分为四条线，如何对应到流水线上呢？如果我们采用同一个代码库管理所有的代码，可以考虑直接将流水线分为四条并行的流程，分别对应这四条线。如果是不同的代码库，则可以考虑对不同的代码库建立不同的流水线。在每条流水线内部，就可以按照单元测试、低集成测试、高集成测试这样的方式组织流水线任务。一、独立的ETL流水线对于ETL代码的测试，有一个值得思考的问题。那就是，ETL脚本之间通常独立性非常强，相互之间没有依赖。这是由于ETL代码常常由完善的领域特定语言SQL开发而成，与Python或Scala等通用编程语言编写的代码不同，SQL文件之间是没有依赖的（如果说有依赖，那也是通过数据库表产生依赖）。既然如此，假设我们修改了某一个ETL文件的代码，是不是我们可以不用运行其他的ETL文件的测试呢？其实不仅如此，我们甚至可以单独上线部署此ETL，而不是一次性部署所有的ETL。这在一定程度上还降低了部署代码带来的风险。有了上面的发现，我们可能要重新思考数据应用的持续交付流水线组织形式。一个可能的办法是为每一个ETL文件建立一个流水线，完成测试、部署的任务。此时每个ETL可以理解为一个独立的小程序。这样的想法在实践中不容易落地，因为这将导致大量的流水线存在（常常有上百条），从而给流水线工具带来了很大的压力。常用的流水线工具，如Jenkins，其设计是难以支撑这么大规模的流水线的创建和管理的。要如何来支持上面这样的ETL流水线呢？可能需要我们开发额外的流水线工具才行。二、云服务中的ETL流水线现在的一些云服务厂商在尝试这样做。他们通常会提供一个基于Web的ETL开发工具，同时会提供工具对当前的ETL的编写测试。此时，ETL开发人员可以在一个地方完成开发、测试、上线，这可以提高开发效率。这类服务的一个常见缺点在于它尝试用一套Web系统来支持所有的ETL开发过程，这带来了大量繁杂的配置。这其实是将ETL开发过程的复杂性转化为了配置的复杂性。相比编写代码而言，多数开发人员不会喜欢这样的工作方式。（当前软件开发所推崇的是Everthing as Code的做法，尝试将所有开发相关过程中的东西代码化，从而可以更好的利用成熟的代码编辑器、版本管理等功能。而Web配置的方式与Everthing as Code背道而驰。）对于这些数据云服务厂商提供的数据开发服务，如果可以同时支持通过代码和Web界面配置来实现数据开发，那将能得到更多开发者的喜爱。这在我看来是一个不错的发展方向。总结由于数据应用开发有很强的独特的特点（比如以SQL为主、有较多的支撑工具等），其测试与功能性软件开发的测试也存在很大的不同。本文分析了如何在测试金字塔的指导下制定测试策略。测试金字塔不仅可以很好的指导功能性软件开发，在进行一般意义上的推广之后，可以很容易得到一般软件的测试策略。关于测试金字塔，本文分析了测试带来的质量信心及测试集成度，这两个概念可以帮助我们更深刻的理解测试金字塔背后的指导原则。在最后，结合我们的实践经验，给出了一些数据应用中的测试构建实践。将数据应用分为四个不同模块来分别构建测试，可以很好的应对数据应用中的质量要求，同时保证有较好的可维护性。最后，我们讨论了如何在持续集成流水线中设计测试任务，留下了一个有待探索的方向，即如何针对单个ETL构建流水线。数据应用的质量保证是不容易做到的，常常需要我们进行很多的权衡取舍才能找到最适合的方式。想要解决这一问题，还要发挥团队中所有人的能动性，多总结和思考才行。文/Thoughtworks 廖光明原文链接：用测试金字塔指导数据应用的测试 - Thoughtworks洞见}