由 Gal Moran、Iris Shmuel 和 Daniel Marcous（Waze 数据科学家）发布

Waze

Waze 是全球最大的基于社区的交通和导航应用程序。它使用实时数据帮助用户绕过字面和比喻意义上的道路颠簸。除了移动导航外，Waze 还提供网络平台、拼车应用程序、合作伙伴服务、广告平台等等。如此广泛的投资组合带来了多种技术挑战和许多不同的用例。

Waze 的机器学习

Waze 依赖于许多机器学习解决方案，包括

• 预测预计到达时间
• 匹配乘客和司机（拼车）
• 投放合适的广告

但要让这些东西变得正确且“生产级”并不容易。对于这些类型的项目来说，通常需要 复杂的周边基础设施 来将它们投入生产，因此需要多个工程师（数据科学家、软件工程师和软件可靠性工程师）以及大量时间。当你将 Waze 的要求（如大规模数据、低（实际上是实时）延迟推理、各种用例以及大量的地理空间数据）考虑在内时，情况更是如此。

以上是为什么机会主义地开始进行机器学习会在 Waze 造成混乱状态的一个很好的理由。对我们来说，它表现为

• 多个机器学习框架 - 你能想到的都有（sklearn、xgboost、TensorFlow、fbprophet、Java PMML、手工制作等）
• 机器学习和操作脱节 - 模型和特征工程嵌入到（Java）后端服务器中，由具有有限监控和验证能力的工程师进行
• 用于训练、验证和部署的半手动操作
• 从想法到生产的漫长的开发周期

总的来说，数据科学家最终将大量时间花在了操作和监控上，而不是专注于实际的建模和数据处理。在一定程度的增长中，我们决定整理混乱并投资于自动化和流程，以便我们可以更快地扩展。我们决定通过采用 完整周期数据科学理念 来大力投资于一种能够显着提高速度和质量的方法。这意味着在这个我们想要建立的新世界中，单个数据科学家能够完成从研究到生产级服务的完整产品周期。

数据科学家现在直接为生产做出贡献，以最大程度地提高影响力。他们专注于建模和数据处理，并且可以开箱即用地获得许多基础设施和操作工作。虽然我们尚未完全实现上述愿景，但我们认为此处提出的努力对于将我们推向正轨至关重要。

Waze 的机器学习堆栈

将上述理念转化为技术规格，我们致力于在 Waze 创建一种简单、稳定、自动化和统一的机器学习管道构建方式。

深入探讨技术要求，我们提出了以下标准

• 简单 - 易于理解、使用、操作
• 托管 - 无服务器，无硬件，只有代码
• 可定制 - 免费获得简单的东西，但足够灵活，可以为需要越界 5% 的情况疯狂
• 可扩展 - 自动可扩展的数据处理、训练、推理
• Pythonic - 我们需要一个生产就绪的东西，它可以使用今天的大多数工具和代码，并且适合标准的数据科学家。如今除了 Python 几乎没有其他选择。

由于上述原因，我们选择了 TFX，以及它内置组件的力量，这些组件可以开箱即用地提供这些功能。

值得一提的是 - Waze 在 Google Cloud Platform (GCP) 上运行其技术堆栈。

碰巧的是，GCP 提供了一套名为 Vertex AI 的工具。它是 Waze 正在其之上构建的机器学习基础设施平台。虽然我们使用了 Vertex AI 托管服务的许多组件，但我们将重点介绍 - Vertex Pipelines - 一个用于机器学习管道的框架，它帮助我们封装 TFX（或任何管道）的复杂性和设置。

连同我们的数据技术堆栈，Waze 的整体机器学习架构（全部托管、可扩展、Pythonic 等）如下所示

仔细阅读的读者会注意到这里有一个明显的警告 - 我们全力投入 TensorFlow。

TFX 意味着 TensorFlow（即使 现在不完全是，让我们假设它就是）。

当你有许多不同的用例时，一开始可能有点吓人。

幸运的是，TF 生态系统很丰富，Waze 有幸拥有足够大的数据，使神经网络可以收敛。

自从开始这项工作以来，我们还没有找到一个 TF 魔法无法更好地或像其他框架一样充分解决的用例（我们不是在谈论微小的百分比，我们不是在进行 Kaggle 竞赛，而是在将东西投入生产）。

Waze TFX

你可能会认为选择 TFX 和 Vertex Pipelines 解决我们所有的问题，但事实并非完全如此。

为了使事情真正变得简单，我们不得不编写一些“粘合代码”（集成上面架构图中的各种产品）并抽象出足够多的细节，以便普通的数据科学家能够有效且快速地使用这些东西。

这导致了

• 消除样板代码
• 隐藏所有常见的 TFX 组件，以便数据科学家只专注于特征工程和建模，并免费获得整个管道
• 生成基于 BigQuery 的训练/评估拆分
• 提供预实现的可选通用特征转换（例如，缩放、归一化、插补）
• 提供预实现的 Keras 模型（例如 DNN/RNN 模型。类似于 TF Estimator，但使用 Keras，可以与 TFX 交谈）
• 实用函数（例如 TF 列准备）
• 用于 tf.transform 特征工程代码的单元测试框架
• 使用安装了所有 TFX 包的 Cloud Run 实例从 Airflow 协调和调度管道运行（无需将其安装在 Airflow Composer 上）

我们将其全部打包在一个易于使用的 Python 包中，名为“waze-data-tfx”

在顶部，我们为数据科学家提供了超级详细的演练、使用指南和代码模板，因此常见的数据科学家工作流程是：分叉、更改配置、稍微调整代码、部署。

作为参考，这是一个简单的 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. 协调

   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,
   )

很简单吧？

当你将一个配置文件提交到代码库时，它会部署并设置持续训练，以及一个完整的管道，包括所有 TFX 和 Vertex AI 的魔力，如数据验证、部署到 Dataflow 的转换、监控等。

总结

当我们的一位数据科学家从长时间休假回来，必须使用这个新的框架来处理一个用例时，我们就知道我们做对了。她说她能够在几个小时内启动一个完整的生产就绪管道，而之前在她休假之前，这需要她几周的时间才能完成。

展望未来，我们计划将许多东西烘焙到 `waze-data-tfx` 中。我们认为拥有这个通用基础设施的一个主要优势是，一旦一个功能添加，每个人都可以“免费”享受它。例如，我们计划向管道添加更多组件，如基础设施验证器和公平指标。一旦这些组件得到支持，每个新的或现有的机器学习管道都会开箱即用地添加这些组件，无需额外的代码。

我们正在计划的其他改进围绕着部署。我们希望在尽可能自动化的情况下提供部署质量保证。

我们目前正在探索的一种方法是使用金丝雀部署。数据科学家只需要配置一个评估指标，框架（使用 Vertex Prediction 的流量拆分功能和其他持续评估魔法）就会在生产中测试新模型，并根据评估的指标逐渐部署或回滚。