目前，网约车已经成为很多人日常出行的首选。对于乘客来说，“我什么时候能到”是下单时最关心的问题；对于平台来说，行程时间预测不仅影响用户体验，更是运营调度、动态定价和司机管理的核心参考依据。
在上一篇文章车联网 | 又快又准的行程时间预测，一个平台搞定？中，我们介绍了如何使用 DolphinDB 精准高效地完成从数据存储到模型预测的全流程。DolphinDB 除了能在离线环境中高效训练机器学习模型，更能在生产环境下实时接入海量订单流数据，完成特征提取与预测，并将结果快速送达前端系统。
今天，我们将结合实例，介绍如何使用 DolphinDB 处理实时订单流数据，基于模型实时预测行程时间，并进行实时监控。
DolphinDB 流数据处理框架
DolphinDB 流数据模块能够满足生产环境对实时数据的高效分析与计算需求。面对服务商不断发送的即时数据，DolphinDB 流数据引擎可以迅速完成数据预处理、信息提取、特征构建等工作，并调用预先训练好的模型，实现对订单行程时间的高效预测，为用户提供涵盖模型训练、流数据接入、实时预测与在线监控的一体化解决方案。
以下是 DolphinDB 的流数据处理框架：
DolphinDB 流数据模块采用“发布-订阅-消费”的模式，流数据首先注入流数据表，通过流表发布数据，第三方应用可以通过 DolphinDB 脚本或 API 订阅及消费流数据。
为实现网约车行程时间的实时预测，服务商可以创建 DolphinDB 流数据表订阅服务端消息，获取乘客创建的行程信息，再使用离线训练完成的模型对行程时间进行实时预测，最后通过应用程序订阅预测数据并提供给乘客。
实时数据模拟及预测
为获取行程数据并使用机器学习模型预测行程时间，用户需要创建三个流表实现实时预测：
- 创建订单信息表订阅乘客行程信息；
- 创建特征表完成对订单信息的特征提取；
- 创建预测表预测特征流数据表发送的行程特征信息，输出预测结果。
流程图如下：
图 1
用户可以使用 subscribeTable 完成流数据的订阅，并通过 handler 指定处理订阅数据的方法。
在本例中，特征表需订阅订单表完成原始信息的特征提取。本模型通过自定义 process 函数对订单数据进行清洗和加工，提取出时间、位置等关键特征。随后，预测表需订阅特征表，并调用自定义的 predictDuration 函数完成行程时间预测，该函数会加载预先训练好的模型，对输入特征进行推理并输出预测结果。
为模拟实时数据，我们使用 replay 函数回放历史数据：
// 订阅订单信息表，数据从订单表流向特征表
subscribeTable(tableName="orderTable", actionName="orderProcess", offset=0, handler=process{traitTable, hisData}, msgAsTable=true, batchSize=1, throttle=1, hash=0, reconnect=true)
// 订阅特征表，数据从特征表流向预测表
subscribeTable(tableName="traitTable", actionName="predict", offset=0, handler=predictDuration{predictTable}, msgAsTable=true, hash=1, reconnect=true)
// 回放历史数据，模拟实时产生的生产数据
submitJob("replay", "trade", replay{inputTables=data, outputTables=orderTable, dateColumn=`pickup_datetime, timeColumn=`pickup_datetime, replayRate=25, absoluteRate=true, parallelLevel=1})Grafana 实时监控
DolphinDB 提供了多种第三方 API 接口，例如 Grafana、Superset、Power BI 等，方便用户进行可视化的分析和监控管理。本文将以 Grafana 为例，介绍如何使用第三方应用程序动态展示实时数据。
DolphinDB 提供了 Grafana 数据接口，用户可以在 Grafana 面板上编写查询脚本与 DolphinDB 进行交互，实现时序数据可视化，并在 Web 端进行实时数据分析。
详见：https://docs.dolphindb.cn/zh/tools/grafana.html
添加 datasource 并新建 dashboard 后，我们可以在 Query 中填写以下 DolphinDB 语句进行实时数据可视化：
- Query 1：展示当日出行订单预估到达时间及预估行程时间
本例中的 predictTable 对应图 1 中的预测表，下同
select id as ID, pickup_datetime as pickup_time, (pickup_datetime+int((exp(duration)-1))) as arrival_time, (exp(duration)-1)/60 as duration from predictTable
where date(predictTable.pickup_datetime) == date(select max(pickup_datetime) from predictTable)- Query 2：统计当日累计订单数及累计乘客数
select count(*) from predictTable
where date(predictTable.pickup_datetime) == date(select max(pickup_datetime) from predictTable)
select sum(passenger_count) from predictTable
where date(predictTable.pickup_datetime) == date(select max(pickup_datetime) from predictTable)
例：订单预估到达时间及当日订单数
- Query 3：统计当日乘客上车位置
select pickup_latitude as latitude, pickup_longitude as longitude from predictTable
where date(predictTable.pickup_datetime) == date(select max(pickup_datetime) from predictTable)- Query 4：统计当日不同时刻订单出行耗时
select pickup_datetime, (exp(duration)-1)/60 as duration from predictTable
where date(predictTable.pickup_datetime) == date(select max(pickup_datetime) from predictTable)数据持久化
订单表保存了最原始、最完整的出行信息，这些数据不仅可以用于与预测结果进行对比分析，评估和优化模型，还能支撑订单量统计、出行规律研究、平台调度与定价策略制定等多种业务需求。因此，可以将订单簿数据落盘存储，用于后续的业务分析。
在 DolphinDB 中，我们可以订阅订单表中数据，指定 subscribeTable 以 loadTable 的方式将数据持久化到磁盘：
db = database("dfs://taxi")
if(existsTable("dfs://taxi", "newData")) { dropTable(db, "newData") }
db.createPartitionedTable(table=table(1:0, orderTable.schema().colDefs.name, orderTable.schema().colDefs.typeString), tableName=`newData, partitionColumns=`pickup_datetime, sortColumns=`pickup_datetime, compressMethods={datetime:"delta"})
subscribeTable(tableName="orderTable", actionName="saveToDisk", offset=0, handler=loadTable("dfs://taxi", "newData"), msgAsTable=true, batchSize=100000, throttle=1, reconnect=true)以上就是使用 DolphinDB 进行网约车行程时间实时预测的全流程。从订单生成到结果输出，DolphinDB 可以为用户提供全链路的低延迟处理和实时的业务决策依据。如果您对 DolphinDB 的更多物联网应用感兴趣，欢迎关注我们，第一时间了解最新动态与技术干货！
)
)
)
)
)
)
)
)
