Waze 是世界上最大的基于社區(qū)的交通和導(dǎo)航應(yīng)用。該應(yīng)用借助實(shí)時數(shù)據(jù)來幫助用戶避開路上的障礙，享受無憂旅途。除了移動導(dǎo)航功能，Waze 還提供網(wǎng)絡(luò)平臺、拼車、合作服務(wù)、廣告投放等功能。如此廣泛的服務(wù)組合帶來了多種技術(shù)挑戰(zhàn)和豐富案例。

Waze 的 ML 應(yīng)用

Waze 依賴許多 ML 解決方案，其中包括：

預(yù)測 ETA

匹配乘客和司機(jī)（拼車）

推送適當(dāng)?shù)膹V告

但是，要把這些事情做好并達(dá)到“生產(chǎn)級別”并不是那么容易。這類項目通常需要復(fù)雜的相關(guān)基礎(chǔ)架構(gòu)，以便將其投入生產(chǎn)，因此需要多位工程師（數(shù)據(jù)科學(xué)家、軟件工程師和軟件可靠性工程師）一起投入大量時間。如果將大規(guī)模數(shù)據(jù)、低延遲（實(shí)際上為實(shí)時）推理、多樣性案例和大量地理空間數(shù)據(jù)等 Waze 的特殊要求結(jié)合在一起，則難度更甚。

復(fù)雜的相關(guān)基礎(chǔ)架構(gòu)

https://papers.nips.cc/paper/5656-hidden-technical-debt-in-machine-learning-systems.pdf

上述理由很好地解釋了，為什么不經(jīng)周密安排就開始在 Waze 中實(shí)施 ML 會造成混亂局面。對我們來說，這表現(xiàn)為：

使用多個 ML 框架——你能想到的都有（sklearn、XGBoost、TensorFlow、fbprophet、Java PMML、人工設(shè)定的框架等等）

ML 和運(yùn)算脫節(jié)——模型和特征工程被工程師嵌入 （Java） 后端服務(wù)器，監(jiān)控和驗(yàn)證能力有限

培訓(xùn)、驗(yàn)證和部署使用半手動操作

從想法到投入生產(chǎn)的開發(fā)周期異常漫長

總的來說，數(shù)據(jù)科學(xué)家最終把大量的時間耗費(fèi)在運(yùn)算和監(jiān)控上，而不能專注于實(shí)際的建模和數(shù)據(jù)處理。

業(yè)務(wù)發(fā)展到一定程度時，我們決定整改這混亂的局面，在自動化和流程方面投資，以便能夠更快地擴(kuò)大業(yè)務(wù)規(guī)模。我們決定采用全周期數(shù)據(jù)科學(xué)理念，重點(diǎn)投資能夠大幅提高速度和質(zhì)量的方法。這意味著，在我們想要建立的這個新方法中，只需一個數(shù)據(jù)科學(xué)家就能完成從研究到生產(chǎn)級服務(wù)的產(chǎn)品周期。

在新方法中，數(shù)據(jù)科學(xué)家可以直接為生產(chǎn)做出貢獻(xiàn)，發(fā)揮自己最大優(yōu)勢。他們可專注于建模和數(shù)據(jù)處理，獲得許多基礎(chǔ)架構(gòu)和開箱即用的運(yùn)算。雖然上述愿景還沒有完全實(shí)現(xiàn)，但我們覺得本文所闡述的內(nèi)容對實(shí)現(xiàn)最終目的來說至關(guān)重要。

Waze 的 ML 堆棧

為將上述理念轉(zhuǎn)化為技術(shù)規(guī)范，我們決心創(chuàng)建一個簡單、穩(wěn)定、自動化和統(tǒng)一的方法來構(gòu)建 Waze 的 ML 流水線。

深入研究技術(shù)要求后，我們得出了以下評判標(biāo)準(zhǔn)：

簡單——易于理解、使用和操作

可托管——沒有服務(wù)器，沒有硬件，只有代碼

可自定義——免費(fèi)獲得簡單的東西，但也有足夠的靈活性，可以為 5% 需要跳出局限的東西而激發(fā)奇思妙想

可擴(kuò)容——自動可擴(kuò)容的數(shù)據(jù)處理、訓(xùn)練、推理

利用 Python——我們需要一些已經(jīng)可以投入生產(chǎn)的工具，可以兼容當(dāng)今大多數(shù)工具和代碼，并且適合一般的數(shù)據(jù)科學(xué)家?，F(xiàn)在除了 Python，幾乎沒有其他選擇

基于上述原因，我們選擇 TFX 及其內(nèi)置組件來提供這些功能，大部分是開箱即用的。

TFX

https://tensorflow.google.cn/tfx

值得一提的是，Waze 是在 Google Cloud Platform （GCP） 上運(yùn)行其技術(shù)棧。

GCP 恰好可提供一套名為 Vertex AI 的工具。Waze 就是在此 ML 基礎(chǔ)架構(gòu)平臺上構(gòu)建的。雖然我們使用了 Vertex AI 托管式服務(wù)的許多組件，但我們將在本文中重點(diǎn)介紹 Vertex Pipelines，這是 ML 流水線的一個框架，能夠幫助我們封裝 TFX（或任何流水線）的復(fù)雜性和設(shè)置。

Vertex Pipelines

https://cloud.google.com/vertex-ai/docs/pipelines

連同我們的數(shù)據(jù)技術(shù)棧，Waze 的整體 ML 架構(gòu)（均符合可托管、可擴(kuò)容，且利用 Python 等標(biāo)準(zhǔn)）如下：

細(xì)心的讀者會在這里注意到所謂的警告：我們所有流程都通過 TensorFlow 進(jìn)行。

TFX 指的是 TensorFlow（盡管這種說法已經(jīng)不再完全準(zhǔn)確了，但我們假設(shè)它是這個意思）。

已經(jīng)不再完全準(zhǔn)確了

https://github.com/tensorflow/tfx/blob/master/tfx/examples/penguin/experimental/penguin_utils_sklearn.py

當(dāng)你處理許多不同的案例時，這些警告一開始可能看起來會有點(diǎn)嚇人。

幸運(yùn)的是，TF 生態(tài)系統(tǒng)很豐富，而且 Waze 的優(yōu)點(diǎn)是有規(guī)模足夠大的數(shù)據(jù)，神經(jīng)網(wǎng)絡(luò)會收斂。

自從開始這項工作以來，我們還沒有發(fā)現(xiàn)在哪個案例中， TF 不能像其他框架那樣更好地或充分地解決問題（在這里不討論微觀的百分點(diǎn)，不進(jìn)行 Kaggle 比賽，而是想得到一些能夠投入生產(chǎn)的東西）。

Waze TFX

您可能認(rèn)為選擇 TFX 和 Vertex 流水線已解決我們所有的問題，但這并不完全正確。

為了讓這項工作真正變得簡單，我們不得不寫一些“膠水代碼”（整合上述架構(gòu)圖中的各種產(chǎn)品），并提取足夠的細(xì)節(jié)，以便普通的數(shù)據(jù)科學(xué)家能夠有效和快速地使用這些東西。

這能帶來以下好處：

可以淘汰樣板文件

可隱藏所有通用的 TFX 組件，這讓數(shù)據(jù)科學(xué)家可以只專注于特征工程和建模，并免費(fèi)獲得整個流水線

可生成基于 BigQuery 的訓(xùn)練/評估分塊

可提供預(yù)先實(shí)現(xiàn)的可選通用特征轉(zhuǎn)換（例如，擴(kuò)容、歸一化、插補(bǔ)）

可提供預(yù)先實(shí)現(xiàn)的 Keras 模型（例如 DNN/RNN 模型。其類似 TF Estimator，但在具有 TFX 特征的 Keras 中）

效用函數(shù)（例如，TF 列的準(zhǔn)備）

tf.transform 特征工程代碼的單元測試框架

使用安裝了所有 TFX 軟件包的云運(yùn)行實(shí)例，從 Airflow 中協(xié)調(diào)和調(diào)度流水線運(yùn)行（無需在 Airflow Composer 上安裝）

我們已經(jīng)將其全部放入一個易于使用的 Python 軟件包中，稱為 “wze-data-tfx”。

在上文中，我們?yōu)閿?shù)據(jù)科學(xué)家提供了一個超級詳細(xì)的演示、使用指南和代碼模板，所以常見的 DS 工作流是：創(chuàng)建分支，改變配置，稍微調(diào)整一下代碼，部署。

作為參考，這里提供了一個簡單的 waze-data-tfx 流水線：

1.配置

_DATASET_NAME = 'tfx_examples'
_TABLE_NAME = 'simple_template_data'

_LABEL_KEY = 'label'
_CATEGORICAL_INT_FEATURES = {
  "categorical_calculated": 2,
}
_DENSE_FLOAT_FEATURE_KEYS = ["numeric_feature1", "numeric_feature2"]
_BUCKET_FEATURES = {
  "numeric_feature1": 5,
}
_VOCAB_FEATURES = {
  "categorical_feature": {
      'top_k': 5,
      'num_oov_buckets': 3
  }
}

_TRAIN_BATCH_SIZE = 128
_EVAL_BATCH_SIZE = 128
_NUM_EPOCHS = 250

_TRAINING_ARGS = {
  'dnn_hidden_units': [6, 3],
  'optimizer': tf.keras.optimizers.Adam,
  'optimizer_kwargs': {
      'learning_rate': 0.01
  },
  'layer_activation': None,
  'metrics': ["Accuracy"]
}

_EVAL_METRIC_SPEC = create_metric_spec([
  mse_metric(upper_bound=25, absolute_change=1),
  accuracy_metric()
])

2.特征工程

def preprocessing_fn(inputs):
  """tf.transform's callback function for preprocessing inputs.

  Args:
      inputs: map from feature keys to raw not-yet-transformedfeatures.

  Returns:
      Map from string feature key to transformed feature operations.
  """
  outputs = features_transform(
      inputs=inputs,
      label_key=_LABEL_KEY,
      dense_features=_DENSE_FLOAT_FEATURE_KEYS,
      vocab_features=_VOCAB_FEATURES,
      bucket_features=_BUCKET_FEATURES,
  )
  return outputs

3.建模

def _build_keras_model(**training_args):
   """Build a keras model.

   Args:
       hidden_units: [int], the layer sizes of the DNN (input layer first).
       learning_rate: [float], learning rate of the Adam optimizer.

   Returns:
       A keras model
   """
   feature_columns = 
       prepare_feature_columns(
           dense_features=_DENSE_FLOAT_FEATURE_KEYS,
           vocab_features=_VOCAB_FEATURES,
           bucket_features=_BUCKET_FEATURES,
       )

   return _dnn_regressor(deep_columns=list(feature_columns.values()),
                         dnn_hidden_units=training_args.get(
                             "dnn_hidden_units"),
                         dense_features=_DENSE_FLOAT_FEATURE_KEYS,
                         vocab_features=_VOCAB_FEATURES,
                         bucket_features=_BUCKET_FEATURES,
                         )

4.協(xié)調(diào)

pipeline_run = WazeTFXPipelineOperator(
   dag=dag,
   task_id='pipeline_run',
   model_name='basic_pipeline_template',
   package=tfx_pipeline_basic,
   pipeline_project_id=EnvConfig.get_value('gcp-project-infra'),
   table_project_id=EnvConfig.get_value('gcp-project-infra'),
   project_utils_filename='utils.py',
   gcp_conn_id=gcp_conn_id,
   enable_pusher=True,
)