DolphinDB 集高性能時(shí)序數(shù)據(jù)庫(kù)與全面的分析功能為一體，可用于海量結(jié)構(gòu)化數(shù)據(jù)的存儲(chǔ)、查詢、分析、實(shí)時(shí)計(jì)算等，在工業(yè)物聯(lián)網(wǎng)場(chǎng)景中應(yīng)用廣泛。本文以紐約出租車行程時(shí)間預(yù)測(cè)為例，介紹如何使用 DolphinDB 訓(xùn)練機(jī)器學(xué)習(xí)模型，并進(jìn)行實(shí)時(shí)數(shù)據(jù)的預(yù)測(cè)，為基于智能網(wǎng)聯(lián)汽車的車聯(lián)網(wǎng)企業(yè)提供基于機(jī)器學(xué)習(xí)方法的即時(shí)預(yù)測(cè)方案。

1. 概要

隨著手機(jī)移動(dòng)應(yīng)用與網(wǎng)約車平臺(tái)的迅速發(fā)展，網(wǎng)約車出行逐漸成為城市生活中一種重要的出行方式。相較其他出行方式而言，選擇網(wǎng)約車的乘客對(duì)出行時(shí)效性有更高的要求，本文將基于乘客上車時(shí)間及上下車地點(diǎn)等靜態(tài)信息，使用 DolphinDB 機(jī)器學(xué)習(xí)方法訓(xùn)練模型，預(yù)測(cè)網(wǎng)約車行程時(shí)間。

在此基礎(chǔ)上，本文將介紹如何使用?DolphinDB 流數(shù)據(jù)處理系統(tǒng)對(duì)業(yè)務(wù)系統(tǒng)產(chǎn)生的持續(xù)增長(zhǎng)的網(wǎng)約車訂單動(dòng)態(tài)數(shù)據(jù)進(jìn)行實(shí)時(shí)的收集、清洗、統(tǒng)計(jì)、入庫(kù)，并實(shí)時(shí)展示行程時(shí)間預(yù)測(cè)結(jié)果。

行程時(shí)間實(shí)時(shí)預(yù)測(cè)流程

2. 數(shù)據(jù)介紹

2.1 數(shù)據(jù)來(lái)源及訓(xùn)練方法

本文訓(xùn)練和預(yù)測(cè)采用?Kaggle?提供的來(lái)自紐約出租車委員會(huì)的數(shù)據(jù)集，訓(xùn)練方法參考了獲獎(jiǎng)?wù)?beluga?的模型，使用 DolphinDB 對(duì)原始數(shù)據(jù)進(jìn)行數(shù)據(jù)預(yù)處理，完成位置信息主成分分析（PCA, Principal Component Analysis）、位置信息聚類（KMeans）、新特征構(gòu)建等工作，并使用 DolphinDB XGBoost 插件完成模型訓(xùn)練及行程時(shí)間預(yù)測(cè)。

為對(duì)比 DolphinDB 在機(jī)器學(xué)習(xí)上的性能，本文使用 Python Scikit-Learn 庫(kù)及 XGBoost 在同一環(huán)境下進(jìn)行了模型訓(xùn)練和預(yù)測(cè)，DolphinDB 在訓(xùn)練耗時(shí)、模型精度等方面均有良好表現(xiàn)。

2.2 數(shù)據(jù)特征

該數(shù)據(jù)集預(yù)先分為訓(xùn)練數(shù)據(jù)集及測(cè)試數(shù)據(jù)集，訓(xùn)練數(shù)據(jù)集共包含1458644條數(shù)據(jù)，測(cè)試數(shù)據(jù)集共包括625134條數(shù)據(jù)；訓(xùn)練數(shù)據(jù)集共包含以下11列信息。

列名	列類型	說(shuō)明	實(shí)例
id	SYMBOL	行程的唯一標(biāo)識(shí)	id2875421
vendor_id	INT	行程記錄提供商代碼	2
pickup_datetime	DATETIME	出租車計(jì)價(jià)器開(kāi)啟時(shí)間	2016/3/14 17:24:55
dropoff_datetime	DATETIME	出租車計(jì)價(jià)器關(guān)閉時(shí)間	2016/3/14 17:32:30
passenger_count	INT	乘客數(shù)量	1
pickup_longitude	DOUBLE	出租車計(jì)價(jià)器開(kāi)啟位置經(jīng)度	-73.98215484619139
pickup_latitude	DOUBLE	出租車計(jì)價(jià)器開(kāi)啟位置緯度	40.76793670654297
dropoff_longitude	DOUBLE	出租車計(jì)價(jià)器關(guān)閉位置經(jīng)度	-73.96463012695312
dropoff_latitude	DOUBLE	出租車計(jì)價(jià)器關(guān)閉位置緯度	40.765602111816406
store_and_fwd_flag	CHAR	標(biāo)識(shí)來(lái)源是否為存儲(chǔ)的歷史數(shù)據(jù)	N
trip_duration	INT	行程時(shí)間（按秒計(jì)）	455

行程時(shí)間預(yù)測(cè)的目標(biāo)列為上表中 trip_duration 列，即 dropoff_datetime 與 pickup_datetime 之差。測(cè)試數(shù)據(jù)集用于預(yù)測(cè)，故其列信息不包括 dropoff_datetime 及 trip_duration 列，測(cè)試數(shù)據(jù)集中行程標(biāo)識(shí)、位置等列屬性同上表。

上表的數(shù)據(jù)類型中，SYMBOL 類型是 DolphinDB 中一種特殊的字符串類型，在系統(tǒng)內(nèi)部的存儲(chǔ)結(jié)構(gòu)為一個(gè)編碼字典，DATETIME 類型為包含了日期和時(shí)刻的時(shí)間類型。

DolphinDB 支持?loadText?方法讀取 csv 等數(shù)據(jù)存儲(chǔ)文件到內(nèi)存表，用戶可以?schema?函數(shù)獲取表的特征信息。DolphinDB 也支持使用 SQL 語(yǔ)句完成數(shù)據(jù)的查詢。

train = loadText("./taxidata/train.csv") 
train.schema().colDefs
select count(*) from train
select top 5 * from train

2.3 數(shù)據(jù)存儲(chǔ)

將數(shù)據(jù)加載到內(nèi)存表后，可以將訓(xùn)練數(shù)據(jù)與測(cè)試數(shù)據(jù)導(dǎo)入 DolphinDB 數(shù)據(jù)庫(kù)中，便于后續(xù)數(shù)據(jù)的讀取與模型的訓(xùn)練，數(shù)據(jù)導(dǎo)入分布式數(shù)據(jù)庫(kù)的操作詳見(jiàn)?database.md · dolphindb/Tutorials_CN - Gitee。

3. 模型構(gòu)建

本節(jié)介紹行程時(shí)間預(yù)測(cè)模型的構(gòu)建方法。

行程時(shí)間預(yù)測(cè)模型的構(gòu)建分多個(gè)過(guò)程，一是預(yù)處理原始數(shù)據(jù)，對(duì)可能存在的空值進(jìn)行轉(zhuǎn)換，并將字符等非數(shù)值型數(shù)據(jù)轉(zhuǎn)換為可用于模型訓(xùn)練的數(shù)值型數(shù)據(jù)；二是優(yōu)化位置信息，原始數(shù)據(jù)中的緯度經(jīng)度信息集中在40.70 °N 至40.80 °N 及73.94 °W 至74.02 °W之間，數(shù)據(jù)間位置特征差異不夠顯著，使用主成分分析、聚類方法處理可以提取到特征更明顯的信息；三是新特征的構(gòu)建，位置信息和時(shí)間信息是訂單數(shù)據(jù)的兩個(gè)關(guān)鍵維度，通過(guò)計(jì)算可以在位置信息基礎(chǔ)上得到方位、距離信息，提取更多空間特征，而組合不同類別的位置信息和時(shí)間信息也可以得到更復(fù)雜的特征，有利于模型學(xué)習(xí)深層次的時(shí)空規(guī)律。

3.1 數(shù)據(jù)預(yù)處理

在模型訓(xùn)練過(guò)程中，首先需要檢查數(shù)據(jù)集是否包含空值，本訓(xùn)練數(shù)據(jù)集與測(cè)試數(shù)據(jù)集均不包含空值，若存在缺失值，還需要?jiǎng)h除、插補(bǔ)等操作解決缺失數(shù)據(jù)問(wèn)題。

其次，需要檢查數(shù)據(jù)集數(shù)據(jù)類型，原始數(shù)據(jù)往往包含文本/字符數(shù)據(jù)，由1.3節(jié)表可知，本數(shù)據(jù)集中 store_and_fwd_flag 列為字符型數(shù)據(jù)，pickup_datetime 及 dropoff_datetime 列為日期時(shí)間類型數(shù)據(jù)，為充分利用這些信息訓(xùn)練模型，需要將其轉(zhuǎn)化為數(shù)值型數(shù)據(jù)。

此外，考慮到該數(shù)據(jù)測(cè)試集評(píng)價(jià)指標(biāo)為均方根對(duì)數(shù)誤差（Root Mean Squared Logarithmic Error, RMSLE），同時(shí)，最大行程時(shí)間接近1000小時(shí)，離群值會(huì)影響模型訓(xùn)練效果，對(duì)行程時(shí)間取對(duì)數(shù)作為預(yù)測(cè)值，在評(píng)價(jià)時(shí)（見(jiàn)3.6節(jié)）可以直接使用均方根誤差（Root Mean Squared Error, RMSE）指標(biāo)。

RMSE

DolphinDB 提供多種計(jì)算函數(shù)，可以幫助用戶快速實(shí)現(xiàn)數(shù)據(jù)處理。DolphinDB 提供?isNull()?方法用于判斷空值，配合?sum()?等聚合函數(shù)使用可以快速完成整表數(shù)據(jù)的查詢；提供類似于條件運(yùn)算符的?iif()?方法簡(jiǎn)化 if-else 語(yǔ)句；date()、weekday()、hour()?等方法可以提取時(shí)間、日期數(shù)據(jù)的不同特征，簡(jiǎn)潔高效；類似于 Python 等編程語(yǔ)言，DolphinDB 支持方括號(hào)([])索引，簡(jiǎn)化了表的查找、更新和插入。

sum(isNull(train))  // 0，不含空值
trainData[`store_and_fwd_flag_int] = iif(trainData[`store_and_fwd_flag] == 'N', int(0), int(1)) // 將字符N/Y轉(zhuǎn)化為0/1值
trainData[`pickup_date] = date(trainData[`pickup_datetime]) // 日期
trainData[`pickup_weekday] = weekday(trainData[`pickup_datetime]) // 星期*
trainData[`pickup_hour] = hour(trainData[`pickup_datetime]) // 小時(shí)
trainData[`log_trip_duration] = log(double(trainData[`trip_duration]) + 1)// 對(duì)行程時(shí)間取對(duì)數(shù)，log(trip_duration+1)
select max(trip_duration / 3600) from trainData // 訓(xùn)練集上最大行程時(shí)間為979h

3.2 位置信息主成分分析（PCA）

原始數(shù)據(jù)中的緯度經(jīng)度信息集中在40.70 °N 至40.80 °N 及73.94 °W 至74.02 °W之間，數(shù)據(jù)間位置特征差異不夠顯著，使用 PCA 來(lái)轉(zhuǎn)換經(jīng)度和緯度坐標(biāo)，有助于 XGBoost 決策樹(shù)的拆分，DolphinDB PCA 函數(shù)使用詳見(jiàn)?pca — DolphinDB 2.0 documentation。

DolphinDB PCA 返回的結(jié)果是一個(gè)字典，包含 components、explainedVarianceRatio、singularValues 三個(gè)鍵，分別代表對(duì)應(yīng)大小為 size(colNames)*k 的主成分分析矩陣、前k個(gè)主成分每個(gè)特征的方差貢獻(xiàn)率、主成分方差（協(xié)方差矩陣特征值）?？赏ㄟ^(guò)主成分分析矩陣轉(zhuǎn)換待處理數(shù)據(jù)，詳見(jiàn)?Scikit-Learn PCA.transform()。

可從中取若干數(shù)據(jù)繪制經(jīng)度-緯度散點(diǎn)圖觀察 PCA 結(jié)果。

經(jīng)處理，位置坐標(biāo)分散在原點(diǎn)附近。

PCA前上客位置信息

PCA 后上客位置信息

pca()?接收一個(gè)或多個(gè)數(shù)據(jù)源為參數(shù)，對(duì)指定列中的數(shù)據(jù)進(jìn)行主成分分析，用戶可通過(guò)?table()?方法創(chuàng)建內(nèi)存表，用于 PCA；DolphinDB 也提供了?dot()、repmat()?等矩陣乘法、矩陣堆疊方法，用戶可使用內(nèi)置函數(shù)快速完成矩陣運(yùn)算，處理位置信息。

PCApara = table(trainData[`pickup_latitude] as latitude, trainData[`pickup_longitude] as longitude)
pca_model = pca(sqlDS(<select * from PCApara>)) // 使用PCA計(jì)算數(shù)據(jù)集屬性
pca_trainpick = dot((matrix(trainPickPara) - repmat(matrix(avg(trainPickPara)), train_n, 1)), pca_model.components) // transform
trainData[`pca_trainpick_0] = flatten(pca_trainpick[:, 0])

DolphinDB 提供了plot函數(shù)供數(shù)據(jù)可視化。用戶可通過(guò)chartType指定圖表類型，詳見(jiàn)plot — DolphinDB 2.0 documentation。

x = select top 1000 pca_trainpick_1 from trainData
y = flatten(matrix(select top 1000 pca_trainpick_0 from trainData))
plot(x, y, chartType=SCATTER)

3.3 位置信息聚類（KMeans）

原始數(shù)據(jù)位置數(shù)據(jù)規(guī)模龐大，很難挖掘多條數(shù)據(jù)間的共同特征。KMeans 可以將經(jīng)緯度相近的數(shù)據(jù)點(diǎn)歸為同一個(gè)簇，有助于更好地歸納組內(nèi)數(shù)據(jù)特征。本模型指定要生成的聚類數(shù)為100，質(zhì)心最大迭代次數(shù)為100，選擇 KMeans++ 算法生成模型，DolphinDB kmeans 可選參數(shù)及含義詳見(jiàn)?kmeans — DolphinDB 2.0 documentation。

可以使用條形圖觀察聚類后的數(shù)據(jù)分布。

KMeans 聚類結(jié)果

kmeans()?接收一個(gè)表作為訓(xùn)練集。對(duì)于機(jī)器學(xué)習(xí)函數(shù)生成的模型，DolphinDB 提供?saveModel?方法將模型保存到本地文件中用于后續(xù)預(yù)測(cè)，用戶可指定服務(wù)器端輸出文件的絕對(duì)路徑或相對(duì)路徑；DolphinDB 也提供了?predict?方法，調(diào)用已訓(xùn)練的特定模型對(duì)相同表結(jié)構(gòu)的測(cè)試集數(shù)據(jù)進(jìn)行預(yù)測(cè)。

kmeans_set = PCApara[rand(size(PCApara)-1, 500000)] // 隨機(jī)選取500000數(shù)據(jù)用于聚類
kmeans_model = kmeans(kmeans_set, 100, maxIter=100, init='k-means++') // KMeans++
saveModel(kmeans_model, "./taxidata/KMeans.model") // 保存模型訓(xùn)練結(jié)果
trainData['pickup_cluster'] = kmeans_model.predict(select pickup_latitude, pickup_longitude from trainData)

saveModel 和 predict 函數(shù)的使用方法可以參考：

• saveModel — DolphinDB 2.0 documentation
• predict — DolphinDB 2.0 documentation

3.4 新特征構(gòu)建

原始數(shù)據(jù)僅提供了經(jīng)度及緯度位置信息，可在此基礎(chǔ)上增加位置特征，如地球表面兩經(jīng)緯度點(diǎn)之間距離、兩經(jīng)緯度點(diǎn)之間的 Manhattan 距離、兩個(gè)經(jīng)緯度之間的方位信息等等。

地球表面兩點(diǎn)間的距離可使用 haversine 公式精確得到，而在本數(shù)據(jù)集中，網(wǎng)約車實(shí)際駛過(guò)的往往是水平或豎直的街道所組成的路徑，Manhattan 距離（也稱城市街區(qū)距離）標(biāo)明兩個(gè)點(diǎn)在標(biāo)準(zhǔn)坐標(biāo)系上的絕對(duì)軸距總和，可能能夠更精確地反映實(shí)際的行駛距離。

在此基礎(chǔ)上，考慮到訓(xùn)練集包含了完整的時(shí)間信息，還可以在訓(xùn)練集上添加速度特征。訓(xùn)練集上的速度特征無(wú)法直接用于測(cè)試集，但在相同的位置聚類屬性或相同的時(shí)間日期特征下，行程所需時(shí)間及平均速度可能存在某些共性（如郊區(qū)或凌晨時(shí)分網(wǎng)約車車速偏大而城區(qū)及早晚高峰時(shí)段網(wǎng)約車車速偏小），可以將這種訓(xùn)練集上發(fā)現(xiàn)的經(jīng)驗(yàn)應(yīng)用到測(cè)試集上?？蓪?shù)據(jù)按聚類屬性或時(shí)間特征分組，統(tǒng)計(jì)組內(nèi)數(shù)據(jù)平均速度等特征，組合生成新特征，合并到測(cè)試集的相應(yīng)分組中。

距離、方位的計(jì)算參數(shù)不同而方法相同，DolphinDB 支持用戶自定義函數(shù)，通過(guò)獨(dú)立的代碼模塊完成特定的計(jì)算任務(wù)。而對(duì)于不同類別（聚類、時(shí)間）內(nèi)的特征，可使用?groupby?方法在每個(gè)分組中計(jì)算需要的特征（如平均值）。groupby?接收三個(gè)參數(shù)，將根據(jù)第三個(gè)參數(shù)指定的列進(jìn)行分組，取第一個(gè)參數(shù)為計(jì)算函數(shù)，計(jì)算第二個(gè)參數(shù)對(duì)應(yīng)列的特征，并返回行數(shù)與分組數(shù)相等的表。用戶可通過(guò)表連接操作將該組合特征合并入特征數(shù)據(jù)，本文使用?fj(full join) 將特征表與 groupby 表合并，fj()?指定第三個(gè)參數(shù)為連接列，將前兩個(gè)參數(shù)所傳入的表合并。

// 兩經(jīng)緯度點(diǎn)距離、兩個(gè)經(jīng)緯度之間的 Manhattan 距離、兩個(gè)經(jīng)緯度之間的方位信息
trainData['distance_haversine'] = haversine_array(trainData['pickup_latitude'], trainData['pickup_longitude'], trainData['dropoff_latitude'], trainData['dropoff_longitude'])
trainData['distance_dummy_manhattan'] = dummy_manhattan_distance(trainData['pickup_latitude'], trainData['pickup_longitude'], trainData['dropoff_latitude'], trainData['dropoff_longitude'])
trainData['direction'] = bearing_array(trainData['pickup_latitude'], trainData['pickup_longitude'], trainData['dropoff_latitude'], trainData['dropoff_longitude'])
// 按時(shí)間、聚類等信息處理速度、行程時(shí)間，產(chǎn)生新特征
for(gby_col in ['pickup_hour', 'pickup_date', 'pickup_week_hour', 'pickup_cluster', 'dropoff_cluster']) {for(gby_para in ['avg_speed_h', 'avg_speed_m', 'log_trip_duration']) {gby = groupby(avg, trainData[gby_para], trainData[gby_col])gby.rename!(`avg_ + gby_para, gby_para + '_gby_' + gby_col)trainData = fj(trainData, gby, gby_col)testData = fj(testData, gby, gby_col)}trainData.dropColumns!(`gby + gby_col)
}

3.5 模型訓(xùn)練（XGBoost）

在進(jìn)行訓(xùn)練之前，需要再一次檢查訓(xùn)練集和測(cè)試集的數(shù)據(jù)，需要剔除 ID、日期、字符等非數(shù)值型數(shù)據(jù)，以及平均速度、行駛時(shí)間等僅在訓(xùn)練集上存在的數(shù)據(jù)，保證訓(xùn)練數(shù)據(jù)與預(yù)測(cè)數(shù)據(jù)結(jié)構(gòu)一致。

完成數(shù)據(jù)處理及特征構(gòu)建后，可以使用 XGBoost 等機(jī)器學(xué)習(xí)方法訓(xùn)練模型。為評(píng)價(jià)模型訓(xùn)練效果，將訓(xùn)練數(shù)據(jù)集劃分為訓(xùn)練集和驗(yàn)證集，隨機(jī)選取80%的數(shù)據(jù)作為訓(xùn)練集訓(xùn)練模型，使用20%的數(shù)據(jù)作為驗(yàn)證集輸出預(yù)測(cè)結(jié)果，使用均方根誤差指標(biāo)計(jì)算驗(yàn)證集的預(yù)測(cè)值與真實(shí)值的偏差。最終可以在測(cè)試集上輸出行程時(shí)間的預(yù)測(cè)結(jié)果。

本模型在驗(yàn)證集上的均方根誤差為0.390，可以繪制預(yù)測(cè)值-真值散點(diǎn)圖，定性分析模型預(yù)測(cè)效果。

驗(yàn)證集上的預(yù)測(cè)值與真實(shí)值

DolphinDB 提供了 XGBoost 插件實(shí)現(xiàn)模型訓(xùn)練及預(yù)測(cè)，使用前需要下載插件到指定路徑并加載 XGBoost 插件。DolphinDB XGBoost 插件使用詳見(jiàn)?xgboost/README_CN.md · dolphindb/DolphinDBPlugin - Gitee。

xgb_pars = {'min_child_weight': 50, 'eta': 0.3, 'colsample_bytree': 0.3, 'max_depth': 10,'subsample': 0.8, 'lambda': 1., 'nthread': 4, 'booster' : 'gbtree', 'silent': 1,'eval_metric': 'rmse', 'objective': 'reg:linear', 'nthread': 48} // xgb 參數(shù)設(shè)置
xgbModel = xgboost::train(ytrain, train, xgb_pars, 60) // 訓(xùn)練模型
yvalid_ = xgboost::predict(xgbModel, valid) // 使用模型進(jìn)行預(yù)測(cè)

3.6 模型評(píng)價(jià)

為實(shí)現(xiàn)出租車行程時(shí)間的預(yù)測(cè)，本文使用了三種機(jī)器學(xué)習(xí)方法。首先使用 PCA 對(duì)位置信息進(jìn)行處理，轉(zhuǎn)換數(shù)據(jù)的經(jīng)緯度特征；使用 KMeans++ 對(duì)出租車上下客位置進(jìn)行聚類，將紐約市區(qū)劃分為100個(gè)區(qū)域進(jìn)行分析；最后使用 XGBoost 對(duì)數(shù)據(jù)集特征進(jìn)行訓(xùn)練。在驗(yàn)證集上模型均方根誤差為0.390，效果較好。

Python Scikit-Learn 也是主流的機(jī)器學(xué)習(xí)庫(kù)之一，本文在相同環(huán)境下使用 Python 對(duì)同一數(shù)據(jù)集進(jìn)行訓(xùn)練，PCA、KMeans++、XGBoost 訓(xùn)練耗時(shí)如下表所示：

模型	DolphinDB	Python
PCA	0.325s	0.396s
KMeans++	45.711s	104.568s
XGBoost	57.269s	74.289s

DolphinDB 與 Python 訓(xùn)練模型在驗(yàn)證集上誤差如下表所示：

	DolphinDB	Python
RMSE	0.390	0.394

在本行程時(shí)間預(yù)測(cè)任務(wù)中，在準(zhǔn)確率上，DolphinDB 與 Python 相近；而在性能上，DolphinDB 在 PCA、KMeans++、XGBoost 上的性能均優(yōu)于 Python。

4. 行程時(shí)間實(shí)時(shí)預(yù)測(cè)

本節(jié)結(jié)合現(xiàn)實(shí)場(chǎng)景，介紹如何使用 DolphinDB 處理實(shí)時(shí)的訂單流數(shù)據(jù)，基于預(yù)測(cè)模型實(shí)時(shí)估計(jì)行程時(shí)間。

現(xiàn)實(shí)場(chǎng)景中，網(wǎng)約車乘客對(duì)時(shí)效性要求高，需要平臺(tái)提供準(zhǔn)確的行程時(shí)間估計(jì)；而服務(wù)商也需要監(jiān)控出行平臺(tái)，分析出行需求并完成資源調(diào)度，僅僅使用預(yù)測(cè)模型無(wú)法高效處理實(shí)時(shí)數(shù)據(jù)，難以完成即時(shí)預(yù)測(cè)任務(wù)，無(wú)法滿足乘客和服務(wù)商的實(shí)時(shí)需求。DolphinDB 流數(shù)據(jù)模塊可以解決生產(chǎn)環(huán)境下實(shí)時(shí)數(shù)據(jù)的快速分析計(jì)算問(wèn)題，對(duì)服務(wù)商發(fā)送的即時(shí)數(shù)據(jù)，DolphinDB 流數(shù)據(jù)引擎可高效完成數(shù)據(jù)預(yù)處理、信息提取、特征構(gòu)建等工作，使用預(yù)先訓(xùn)練的模型完成實(shí)時(shí)訂單行程時(shí)間的快速準(zhǔn)確預(yù)測(cè)，為用戶提供從模型訓(xùn)練、流數(shù)據(jù)注入到實(shí)時(shí)預(yù)測(cè)及在線監(jiān)控的一站式解決方案。

DolphinDB 流數(shù)據(jù)處理框架

4.1 場(chǎng)景描述

DolphinDB 流數(shù)據(jù)模塊采用發(fā)布-訂閱-消費(fèi)的模式，流數(shù)據(jù)首先注入流數(shù)據(jù)表中，通過(guò)流表來(lái)發(fā)布數(shù)據(jù)，第三方應(yīng)用可以通過(guò) DolphinDB 腳本或 API 訂閱及消費(fèi)流數(shù)據(jù)。

為實(shí)現(xiàn)出租車行程時(shí)間的實(shí)時(shí)預(yù)測(cè)，服務(wù)商可以創(chuàng)建 DolphinDB 流數(shù)據(jù)表訂閱服務(wù)端消息，獲取乘客創(chuàng)建的行程信息，使用離線訓(xùn)練完成的模型對(duì)行程時(shí)間進(jìn)行實(shí)時(shí)預(yù)測(cè)，最后可通過(guò)應(yīng)用程序訂閱預(yù)測(cè)數(shù)據(jù)并提供給乘客。

4.2 實(shí)時(shí)數(shù)據(jù)模擬及預(yù)測(cè)

為獲取行程數(shù)據(jù)并使用機(jī)器學(xué)習(xí)模型預(yù)測(cè)行程時(shí)間，用戶需要?jiǎng)?chuàng)建三個(gè)流表實(shí)現(xiàn)實(shí)時(shí)預(yù)測(cè)，一是創(chuàng)建訂單信息表訂閱乘客行程信息，二是創(chuàng)建特征表完成對(duì)訂單信息的特征提取；三是創(chuàng)建預(yù)測(cè)表預(yù)測(cè)特征流數(shù)據(jù)表發(fā)送的行程特征信息，輸出預(yù)測(cè)結(jié)果。

用戶可以使用 subscribeTable 完成流數(shù)據(jù)的訂閱，并通過(guò)?handler?指定處理訂閱數(shù)據(jù)的方法（詳見(jiàn)subscribeTable — DolphinDB 2.0 documentation）。在本例中，特征表需訂閱訂單表完成原始信息的特征提取，本模型定義?process?函數(shù)實(shí)現(xiàn)；預(yù)測(cè)表需訂閱特征表使用特征信息完成行程時(shí)間預(yù)測(cè)，本模型定義?predictDuration?函數(shù)實(shí)現(xiàn)。函數(shù)實(shí)現(xiàn)詳見(jiàn)6.2節(jié)所附代碼。

為模擬實(shí)時(shí)數(shù)據(jù)，使用?replay?函數(shù)回放歷史數(shù)據(jù)。

// 訂閱訂單信息表，數(shù)據(jù)從訂單表流向特征表
subscribeTable(tableName="orderTable", actionName="orderProcess", offset=0, handler=process{traitTable, hisData}, msgAsTable=true, batchSize=1, throttle=1, hash=0, reconnect=true)
// 訂閱特征表，數(shù)據(jù)從特征表流向預(yù)測(cè)表
subscribeTable(tableName="traitTable", actionName="predict", offset=0, handler=predictDuration{predictTable}, msgAsTable=true, hash=1, reconnect=true)
// 回放歷史數(shù)據(jù)，模擬實(shí)時(shí)產(chǎn)生的生產(chǎn)數(shù)據(jù)
submitJob("replay", "trade",  replay{inputTables=data, outputTables=orderTable, dateColumn=`pickup_datetime, timeColumn=`pickup_datetime, replayRate=25, absoluteRate=true, parallelLevel=1})

4.3 Grafana 實(shí)時(shí)監(jiān)控

服務(wù)商可通過(guò)第三方 API 連接 DolphinDB 數(shù)據(jù)庫(kù)監(jiān)控行程時(shí)間預(yù)測(cè)服務(wù)，本文以 Grafana 為例簡(jiǎn)要介紹如何使用第三方應(yīng)用程序動(dòng)態(tài)展示實(shí)時(shí)數(shù)據(jù)。

Grafana 是一個(gè)用于時(shí)序數(shù)據(jù)動(dòng)態(tài)可視化的數(shù)據(jù)展示工具，DolphinDB 提供了 Grafana 的數(shù)據(jù)接口，用戶可在 Grafana 面板上編寫(xiě)查詢腳本與 DolphinDB 進(jìn)行交互，實(shí)現(xiàn) DolphinDB 時(shí)序數(shù)據(jù)的可視化，并 Web 端進(jìn)行實(shí)時(shí)數(shù)據(jù)分析，詳見(jiàn)?README.zh.md · dolphindb/grafana-datasource - Gitee。

添加 datasource 并新建 dashboard 后，在 Query 中填寫(xiě)以下 DolphinDB 語(yǔ)句進(jìn)行實(shí)時(shí)數(shù)據(jù)可視化：

• Query 1：展示當(dāng)日出行訂單預(yù)估到達(dá)時(shí)間及預(yù)估行程時(shí)間

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：統(tǒng)計(jì)當(dāng)日累計(jì)訂單數(shù)及累計(jì)乘客數(shù)

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) 

訂單預(yù)估到達(dá)時(shí)間及當(dāng)日訂單數(shù)

• Query 3：統(tǒng)計(jì)當(dāng)日乘客上車位置

select pickup_latitude as latitude, pickup_longitude as longitude from predictTable 
where date(predictTable.pickup_datetime) == date(select max(pickup_datetime) from predictTable) 

當(dāng)日乘客上車位置

• Query 4：統(tǒng)計(jì)當(dāng)日不同時(shí)刻訂單出行耗時(shí)

select pickup_datetime, (exp(duration)-1)/60 as duration from predictTable 
where date(predictTable.pickup_datetime) == date(select max(pickup_datetime) from predictTable) 

當(dāng)日不同時(shí)刻訂單出行耗時(shí)

4.4 數(shù)據(jù)持久化

如果需要將歷史數(shù)據(jù)落盤，可以訂閱訂單表中數(shù)據(jù)，指定?subscribeTable?以?loadTable?的方式將數(shù)據(jù)持久化到磁盤。

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)

5. 總結(jié)

本文介紹了使用 DolphinDB 機(jī)器學(xué)習(xí)函數(shù)及插件訓(xùn)練出租車行程時(shí)間預(yù)測(cè)模型的方法，與 Python Scikit-Learn 等主流機(jī)器學(xué)習(xí)方法相比，DolphinDB 在模型訓(xùn)練耗時(shí)及預(yù)測(cè)精度上均有良好表現(xiàn)；在此基礎(chǔ)上，本文還介紹了如何使用 DolphinDB 流數(shù)據(jù)處理工具進(jìn)行實(shí)時(shí)預(yù)測(cè)，并以 Grafana 為例展示了 DolphinDB 時(shí)序數(shù)據(jù)的可視化方法。DolphinDB 內(nèi)置的計(jì)算函數(shù)和機(jī)器學(xué)習(xí)方法能夠?qū)崿F(xiàn)從數(shù)據(jù)存儲(chǔ)、數(shù)據(jù)載入、數(shù)據(jù)清洗、特征構(gòu)造到模型建立、模型評(píng)價(jià)的完整機(jī)器學(xué)習(xí)流程，可以為物聯(lián)網(wǎng)行業(yè)用戶提供更全面的數(shù)據(jù)分析方法。

6. 附錄

6.1 測(cè)試環(huán)境

• 操作系統(tǒng)：Linux version 3.10.0-1160.el7.x86_64
• CPU：Intel(R) Xeon(R) Silver 4214 CPU @2.20GHz 48核
• 內(nèi)存：188G
• 軟件版本：
  • DolphinDB：2.00.9
  • Python3：3.7.12
  • Scikit-Learn：1.0.2
  • XGBoost：1.6.2

6.2 模型代碼

DolphinDB 模型訓(xùn)練代碼：taxiTrain.dos

DolphinDB 流數(shù)據(jù)預(yù)測(cè)代碼：?taxiStream.dos

Python 模型訓(xùn)練代碼：?taxiTrain.py