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

Waze 的 ML 應用

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

預測 ETA

匹配乘客和司機（拼車）

推送適當?shù)膹V告

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

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

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

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

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

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

培訓、驗證和部署使用半手動操作

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

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

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

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

Waze 的 ML 堆棧

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

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

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

可托管——沒有服務器，沒有硬件，只有代碼

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

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

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

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

TFX

https://tensorflow.google.cn/tfx

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

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

Vertex Pipelines

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

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

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

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

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

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

當你處理許多不同的案例時，這些警告一開始可能看起來會有點嚇人。

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

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

Waze TFX

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

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

這能帶來以下好處：

可以淘汰樣板文件

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

可生成基于 BigQuery 的訓練/評估分塊

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

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

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

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

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

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

在上文中，我們?yōu)閿?shù)據(jù)科學家提供了一個超級詳細的演示、使用指南和代碼模板，所以常見的 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,
)