X-Deep Learning功能模塊
? 特征體系
o 樣本
o 特征
o 網絡
? 數據準備
o 樣本格式
o 使用DataReader讀取數據
o 自定義python reader
? 定義模型
o 稀疏部分
o 稠密部分
o 優化器
? 訓練模型
o 單機訓練
o 分布式訓練
o 同步及半同步訓練
o ID準入和ID退出
o 保存及恢復模型變量
? 模型評估
o 模型評估
? 高層訓練API
o Estimator
? 調試
o Timeline

1. 特征體系
   1.1 樣本
   樣本由 label, feature, sampleid 三部分組成；訓練過程是通過feature學習label的過程，sample id是一條樣本的標識。label可以是單值(二分類)或者多值(多分類)。
   1.2 特征
   傳統的特征一般是稠密的(dense)，維度不會特別高，樣本都出現；以向量的形式表示；例如圖像特征，或者用戶的性別信息。
   對于搜索-推薦-廣告場景，存在大量的稀疏特征(sparse)：這些特征維度高（百億），但是樣本中出現次數低（數百)，這類特征以多個kv的方式稀疏表示；例如商品類目(key是類目的id，沒有value)，用戶點擊過的商品列表(key是商品id，value是點擊次數)。
   特征以特征組的方式組織，便于構建訓練網絡；例如特征分為：用戶性別(稠密特征)，用戶偏好(稀疏特征)，商品類目(稀疏特征)，商品價格(稠密特征)；
   1.3 網絡
   簡單的稀疏場景訓練網絡
2. 數據準備
   2.1 樣本格式
   2.1.1 文本格式
   文本格式一行表示一條樣本，分為多個字段，用’|'分隔 字段定義和分隔符如下：
   

7859345_420968_1007|user_3423487|clk_14@32490:1.0,32988:2.0;prefer@323423,32342|qscore@0.8,0.5;ad_price@33.8|0.0,1.0|1544094136
2.1.2 protobuf格式
簡單樣本格式用message定義一條樣本，包括 sample_id, label, feature_line 三個部分； 文件中先以一個32位的整數記錄樣本長度，然后存儲序列化后的樣本；
// 簡單樣本格式
message SampleSimple {
required string sample_id = 1; // 樣本的sample_id
required Label label = 2; // 樣本的label， Label類型
required FeatureLine feature_line = 3; // 樣本的特征行
}
? 更多內容見結構化壓縮
2.2 使用DataReader讀取數據
2.2.1 DataReader Api
import xdl

reader = xdl.DataReader(“r1”, # reader名稱
paths=["./data.txt"], # 文件列表
enable_state=False) # 是否打開reader state，用于分布式failover，開啟的時候需要額外的命令行參數(task_num)

reader.epochs(1).threads(1).batch_size(10).label_count(1)
reader.feature(name=‘sparse0’, type=xdl.features.sparse)\ # 定義reader需要讀取的特征，本例包括兩個sparse特征組和一個dense特征組
.feature(name=‘sparse1’, type=xdl.features.sparse)
.feature(name=‘deep0’, type=xdl.features.dense, nvec=256)
reader.startup()

…

2.2.2 選項設置

? 更多選項見更多選項
2.2.3 結構化壓縮
結構化壓縮是指，多個樣本中共同的特征值，只存儲一份，也只進行一次計算；需要在樣本處理中，聚合多條具有共同特征的樣本。
例如廣告樣本中，一個用戶可能點擊N個廣告，產生N條樣本；這N條樣本的廣告特征簇都是不同的，但是用戶特征簇都是相同的一份。
使用結構化壓縮，需要在pb樣本中表達特征結構化，主要是通過定義多個特征表
例如下圖定義了一個特征主表(ad)和一個特征輔表(user)，通過主表的多個ad特征指向輔表的一個user特征，表示多個樣本的特征復用關系。
a0 和 a1 特征共用一個u0特征；a2, a3, a4共用一個u1特征
特征表的定義是一個repeated， 這樣可以通過定義兩個以上的特征表，來表示多層壓縮。例如 圖片 -> 廣告 -> 用戶 這樣兩層的多對一關系
message SampleGroup {
repeated string sample_ids = 1; // 每個樣本的sample_id
repeated Label labels = 2; // 每個樣本的label， Label類型
repeated FeatureTable feature_tables = 3; // 整個sample的特征表，如果沒有輔表，只有一個feature_table
repeated Extensions extensions = 5; // 每個樣本的擴展字段，待以后擴展

}

message Extensions {
map<string, string> extension = 1;
}

// 標簽，支持多目標訓練
message Label {
repeated float values = 1;
}

// 特征表
message FeatureTable {
repeated FeatureLine feature_lines = 1; // 每個樣本的特征行
}

// 特征行
message FeatureLine {
repeated Feature features = 1; // 每個特征行里的特征(組)
optional int32 refer = 2; // 引用下層輔表的哪個特征行
}

// 特征(組)
message Feature {
required FeatureType type = 1; // 特征類型
optional string name = 2; // 特征(組)名字，與field_id二選一
repeated FeatureValue values = 3; // 特征值, 一個特征(組)可能有多個特征值
}

// 特征值
message FeatureValue {
optional int64 key = 1; // 特征ID, dense可以沒有
optional float value = 2; // 特征值，沒有默認是1
repeated float vector = 3; // 特征向量，向量類型的特征才有，也可以用來表示稠密特征
optional int64 hkey = 4; // 特征ID高64位，用來支持128位hashkey
}
2.2.4 更多選項

2.3 自定義python reader
? xdl支持直接使用python定義op
from tensorflow.examples.tutorials.mnist import input_data

mnist_data = input_data.read_data_sets(’./data’)

python讀取函數，直接使用tf封裝好的api讀取mnist數據

def read_data(batch_size=100):
global mnist_data
images, labels = mnist_data.train.next_batch(batch_size)
labels = np.asarray(labels, np.float32)
return images, labels

通過xdl.py_func定義op

images, labels = xdl.py_func(read_data, [], output_type=[np.float32, np.float32])

3. 定義模型
   XDL專注解決搜索廣告等稀疏場景的模型訓練性能問題，因此將模型計算分為稀疏和稠密兩部分，稀疏部分通過參數服務器，GPU加速，參數合并等技術極大提升了稀疏特征的計算和通信性能。稠密部分采用多backend設計，支持TF和Mxnet兩個引擎作為計算后端，并且可以使用原生TF和Mxnet API定義模型。下面分別介紹稀疏和稠密部分的API
   3.1 稀疏API
   ? API列表
   

? 參數說明
? xdl.embedding

xdl.merged_embedding

3.2 稠密API
XDL使用TF和Mxnet作為計算后端，并且支持使用TF和Mxnet原生API來定義模型
? 定義方法
a. 使用TensorFlow或者Mxnet定義模型
b. 使用xdl.tf_wrapper或者xdl.mxnet_wrapper修飾模型定義函數
? 裝飾器參數
?

? 使用TF Backend定義一個embedding + 5層dense網絡
@xdl.tf_wrapper()
def model_fn(dense, embeddings, labels):
input_features = [dense]
input_features.extend(embeddings)
inputs = tf.concat(input_features, 1)
fc1 = tf.layers.dense(inputs, 256, activation=tf.nn.relu)
fc2 = tf.layers.dense(fc1, 128, activation=tf.nn.relu)
fc3 = tf.layers.dense(fc2, 64, activation=tf.nn.relu)
fc4 = tf.layers.dense(fc3, 32, activation=tf.nn.relu)
logits = tf.layers.dense(fc4, 1, activation=tf.nn.relu)
cross_entropy = tf.nn.sigmoid_cross_entropy_with_logits(labels=labels, logits=logits)
loss = tf.reduce_mean(cross_entropy)
return loss
? 使用Mxnet Backend定義一個embedding + 5層dense網絡
@xdl.mxnet_wrapper()
def model_fn(dense, embeddings, label):
input_features = [dense]
input_features.extend(embeddings)
inputs = mx.symbol.concat(*input_features, dim=1)
fc1 = mx.sym.FullyConnected(data=inputs, num_hidden=256, name=‘fc1’)
fc2 = mx.sym.FullyConnected(data=fc1, num_hidden=128, name=‘fc2’)
fc3 = mx.sym.FullyConnected(data=fc2, num_hidden=64, name=‘fc3’)
fc4 = mx.sym.FullyConnected(data=fc3, num_hidden=32, name=‘fc4’)
fc5 = mx.sym.FullyConnected(data=fc4, num_hidden=1, name=‘fc5’)
prop = mx.symbol.SoftmaxOutput(data=fc5, label=label)
loss = - mx.symbol.sum(mx.symbol.log(prop) * label) / 4
return loss
3.3 優化器
? XDL支持常用的optimizer，包括
o SGD
o Momentum
o Agagrad
o Adam
o Ftrl
? 使用方法
optimizer = xdl.SGD(0.5)
train_op = optimizer.optimize()
sess = xdl.TrainSession()
sess.run(train_op)
4. 訓練模型
XDL支持單機及分布式兩種訓練模式，單機模式一般用來做早期模型的調試和正確性驗證，為了充分發揮XDL的稀疏計算能力，建議使用分布式模式進行大規模并行訓練
4.1 單機訓練
XDL通過Local PS的方式支持單機訓練，只需運行時給python腳本加上–run_mode=local的命令行參數即可：
python test.py --run_mode=local
如果用戶需要使用XDL鏡像進行單機訓練，則需要先以bash session形式進入鏡像，再啟動命令：
sudo docker run -it xdl/xdl:tag /bin/bash
python test.py --run_mode=local
4.2 分布式訓練
? XDL通過ams存儲參數，從而支持了分布式訓練，在進行分布式訓練時，需要啟動ams進程和worker進程，ams進程包括一個scheduler和多個server，用戶可以通過手動方式啟動，也可以使用XDL提供的基于yarn+docker的分布式調度工具xdl_sumbit啟動
4.2.1 通過手工方式啟動分布式任務
以下命令都默認宿主機上具有XDL運行環境，或用戶以bash session形式進入XDL鏡像。
? 啟動ams-scheudler

參數解釋：

ps_cpu_cores和ps_memory_m：分給ams-server的cpu和內存資源，將會影響ams的參數分配算法
ckpt_dir：checkpoint地址，目前支持本地和HDFS兩種文件系統

python test.py --task_name=scheduler --zk_addr=zfs://xxx --ps_num=2 --ps_cpu_cores=10 --ps_memory_m=4000 --ckpt_dir=hdfs://xxx/checkpoint
? 啟動ams-server
python test.py --task_name=ps --task_index=0 --zk_addr=zfs://[zk_server_ip:port]
python test.py --task_name=ps --task_index=1 --zk_addr=zfs://[zk_server_ip:port]
? 啟動worker
python test.py --task_name=worker --task_index=0 --task_num=4 --zk_addr=zfs://[zk_server_ip:port]
python test.py --task_name=worker --task_index=1 --task_num=4 --zk_addr=zfs://[zk_server_ip:port]
python test.py --task_name=worker --task_index=2 --task_num=4 --zk_addr=zfs://[zk_server_ip:port]
python test.py --task_name=worker --task_index=3 --task_num=4 --zk_addr=zfs://[zk_server_ip:port]
4.2.2 通過xdl_submit啟動分布式任務
? 使用xdl_submit需要在機器上提前部署相關環境，部署方法參見集群部署
? xdl_submit任務基礎配置示例
{
“job_name”: “xdl_test”,
“dependent_dirs”: “/home/xdl_user/xdl_test/”,
“script”: “test.py”,
“docker_image”: “registry.cn-hangzhou.aliyuncs.com/xdl/xdl:ubuntu-cpu-tf1.12”,
“worker”: {
“instance_num”: 10,
“cpu_cores”: 4,
“gpu_cores”: 0,
“memory_m”: 1000
},
“ps”: {
“instance_num”: 2,
“cpu_cores”: 2,
“gpu_cores”: 0,
“memory_m”: 1000
},
“checkpoint”: {
“output_dir”: “hdfs://ns1/data/xdl_user/xdl_test/checkpoint”
}
}
? 基礎配置項說明 （帶*號為必要配置項）
?

? xdl_submit任務高級配置示例
{
“job_name”: “xdl_test”,
“dependent_dirs”: “/home/xdl_user/xdl_test/”,
“script”: “test.py”,
“docker_image”: “registry.cn-hangzhou.aliyuncs.com/xdl/xdl:ubuntu-cpu-tf1.12”,
“worker”: {
“instance_num”: 10,
“cpu_cores”: 4,
“gpu_cores”: 0,
“memory_m”: 1000
},
“ps”: {
“instance_num”: 2,
“cpu_cores”: 2,
“gpu_cores”: 0,
“memory_m”: 1000
},
“checkpoint”: {
“output_dir”: “hdfs://ns1/data/xdl_user/xdl_test/checkpoint”
},
“scheduler_queue”: “default”,
“min_finish_worker_rate”: 90,
“max_failover_times”: 20,
“max_local_failover_times”: 3,
“extend_role”:{
“ams”: {
“instance_num”: 10,
“cpu_cores”: 8,
“gpu_cores”: 0,
“memory_m”: 8000,
“script”: “ams.py”
}
}
}
? 高級配置項說明

? 提交任務
o 將test.py放到/home/xdl_user/xdl_test/目錄下，如果有其他腳本或者本地數據和配置也可以放到該目錄下，xdl_submit會自動將其掛載到docker內
o 執行命令: xdl_submit.py --config=xdl_test.json
4.3 同步及半同步訓練
? 同步訓練
#創建session時，添加同步訓練的hook
hooks = []
hooks.append(xdl.SyncRunHook(xdl.get_task_index(), xdl.get_task_num()))
sess = xdl.TrainSession(hooks)
while not sess.should_stop():
sess.run(train_ops)
#sess run結束后，需要調用worker_report_finish_op
xdl.worker_report_finish_op(np.array(xdl.get_task_index(),dtype=np.int32))

? 半同步訓練
#創建session時，添加半同步訓練的hook，staleness為不同worker間允許的最大差異step數，默認值為0
hooks = []
hooks.append(xdl.SemiSyncRunHook(xdl.get_task_index(), xdl.get_task_num(), staleness=0))
sess = xdl.TrainSession(hooks)
while not sess.should_stop():
sess.run(train_ops)
#sess run結束后，需要調用worker_report_finish_op
xdl.worker_report_finish_op(np.array(xdl.get_task_index(),dtype=np.int32))
4.4 ID準入和ID退出
? 基于概率的ID準入
XDL在xdl.embedding接口中提供了基于概率的ID準入，使用方法如下：
emb1 = xdl.embedding(‘emb1’, batch[‘sparse0’], xdl.TruncatedNormal(stddev=0.001), 8, 1024, vtype=‘hash’, feature_add_probability=0.9)
feature_add_probability表示ID準入的概率，0表示哈希表完全不接受新的ID，1表示所有ID都會準入到哈希表。
? 基于更新時間的ID退出
如果某個ID長時間沒被更新，那說明這個ID在模型中已經處于不太重要的地位，XDL提供了刪除這些ID的功能，使用方法如下：
emb1 = xdl.embedding(‘emb1’, batch[‘sparse0’], xdl.TruncatedNormal(stddev=0.001), 8, 1024, vtype=‘hash’)
emb2 = xdl.embedding(‘emb2’, batch[‘sparse1’], xdl.TruncatedNormal(stddev=0.001), 8, 1024, vtype=‘hash’)

hooks = []

vars = [“emb1”, “emb2”]
mark_hook1 = xdl.GlobalStepMarkHook(“emb1”, batch[“sparse0”].ids)
mark_hook2 = xdl.GlobalStepMarkHook(“emb2”, batch[“sparse1”].ids)
hooks.append(mark_hook1)
hooks.append(mark_hook2)
if xdl.get_task_index() == 0:
filter_hook = xdl.GlobalStepFilterHook(vars, 30, 10)
hooks.append(filter_hook)
方法說明：

1. "emb1"和"emb2"是兩路需要進行ID退出的特征，針對這兩路特征創建兩個GlobalStepMarkHook，具體創建方法參見上述代碼；
2. 選取一個worker（這里選擇worker0），創建一個GlobalStepFilterHook，第一個參數"vars"為需要進行ID退出的變量名稱集合，第二個參數"30"表示global_step每隔30步，進行一次ID退出的動作，第三個參數"10",表示如果某個ID超過10步沒有被更新，在下一次的ID退出動作時，這個ID就會被刪除。
   4.5 保存和恢復模型變量
   ? 保存模型變量
3. 通過Saver保存
   import xdl
   saver = xdl.Saver()
   checkpoint_version = “xxx” # checkpoint名稱，一般是global step
   saver.save(version = checkpoint_version)
4. 通過CheckpointHook保存
   import xdl
   train_op = …
   hook = xdl.CheckpointHook(save_interval_step=1000) # 每1000個global step保存一次
   sess = xdl.TrainSession(hooks=[hook])
   sess.run(train_op)
   ? 恢復模型參數
   import xdl
   saver = xdl.Saver()
   checkpoint_version = “xxx”
   saver.restore(version = checkpoint_version)
   4.5 將二進制模型轉為可讀的文本格式
   import xdl

variables=[“x1”, “x2”] # list of variable names
xdl.convert_ps_variable(ckpt_dir=“hdfs://absolute_path/to/checkpoint_dir”,
output_dir=“hdfs://absolute_path/to/output_dir”, var_list=variables)
ckpt_dir既可以指定到checkpoint目錄也可以指定到具體版本(ckpt…xxxx這一級)。指定到checkpoint目錄會自動找到最新版本
5. 模型評估
模型評估是用指標反映模型在實際數據中的表現，是在訓練中調整超參數，評估模型效果的重要依據。XDL提供了計算auc的默認op實現，用戶也可以通過python或者c++定制自己的metrics實現
import xdl
saver = xdl.Saver()
saver.restore(ckpt_version)
labels = …
predictions = …
auc = xdl.auc(predictions, labels)
sess = xdl.TrainSession()
print sess.run(auc)
6. 高層訓練API：Estimator
為了簡化用戶編寫模型訓練腳本的工作量，XDL提供了Estimator API，可以允許用戶以一套代碼執行訓練/預測/評估/訓練&評估等多種類型的任務
使用步驟
? 定義輸入function

定義train輸入

def input_fn():
…
return feature_list, labels

定義predict/evaluate輸入

def eval_input_fn():
…
return test_feature_list, test_labels
? 定義模型
@xdl.tf_wrapper()
def model_fn(feature_list, labels):
logits = …
loss = …
return loss, logits
? 創建Estimator
estimator = xdl.Estimator(model_fn=model_fn, optimizer=xdl.SGD(0.5))
? 進行train|evaluate|predict|train&&evaluate

訓練

estimator.train(input_fn, max_step=2000, checkpoint_interval=1000)

評估: checkpoint_version=""表示從最后一個checkpoint讀取參數

estimator.evaluate(eval_input_fn, checkpoint_version="", max_step=2000)

預測

estimator.predict(eval_input_fn, checkpoint_version="", max_step=2000)

訓練和評估交替執行

estimator.train_and_evaluate(train_input_fn=input_fn,
eval_input_fn=eval_input_fn,
eval_interval=1000,
eval_steps=200,
checkpoint_interval=1000,
max_step=5000)
7. Timeline
使用步驟
? 在訓練中產出timeline
run_option = xdl.RunOption()
run_option.perf = True
run_statistic = xdl.RunStatistic()
_ = sess.run(train_ops, run_option, run_statistic)
xdl.Timeline(run_statistic.perf_result).save(’./timeline.json’)

? 在chrome中輸入chrome://tracing，加載timeline.json

總結

以上是生活随笔為你收集整理的X-Deep Learning功能模块的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。