TFX MLOps 0.5天快速上手-Kubeflow版

AI / ML / DL专业的学生，拿到老师/助教整理好的数据，在colab/notebook上构建模型，爆调一堆参数，模型达标后交完作业/论文就结束了。 在产业界可不这么容易! 首先，数据是野生的raw data； 单机notebook/colab也不足以训练稍大型的模型；模型也需要定期/不定期重新使用新数据训练/调参和重新部署，全靠人工来完成成本可能会高到老板不会同意。

MLOps认为ML是一个循环渐进的过程，每一次循环，模型性能都应该有所提升

学校/实验室里的ML V.S. 产业界的ML

本文介绍如何使用TFX把机器学习项目中数据接入/处理、特性工程、模型训练与评估、模型推导性能验证（速度与能耗是否达标）、多端部署整个流程自动化/工业标准化/规模化的， 带着这个问题思路watch和read文中的视频和链接到教程就能懂（PS.单看本文是不会懂的，需要按顺序看文中的视频和教程，该动手就动动小手）。

网上的教程两极分化，一种是谈天说地，天花乱坠地吹了半天TFX多么多么强大，但却没有演示一个小小例子让读者感受一下用起来是怎么样的； 另一种直接扔了一连串的notebook代码+GCP图形界面+命令行操作，没头没尾没有上下文的，看完也不知道怎么构建第一个Pipeline。所以才有了本文。

很多教程都从TFX本身的架构开始……，其实初学者可以先不懂这块

那么如何在半天的时间里上手TFX MLOps呢？

第一步： 听笔者用平实的语言废话10秒钟

MLOps类似DevOPS，后者将软件开发项目构建、QA、部署自动化；前者将机器学习项目自动化。TFX是Google提供的，用于实现此目标的工具组件。

Kubeflow呢，Kubeflow is an open source ML platform dedicated to making deployments of machine learning (ML) workflows on Kubernetes simple, portable and scalable. Kubeflow Pipelines is part of the Kubeflow platform that enables composition and execution of reproducible workflows on Kubeflow, integrated with experimentation and notebook based experiences. Kubeflow Pipelines services on Kubernetes include:
- the hosted Metadata store,
- container based orchestration engine,
- notebook server,
- and UI to help users develop, run, and manage complex ML pipelines at scale.

第三步中的tfx Pipeline 不一定要跑在Kubeflow上， 其它编排器 Apache Airflow、 Apache Beam也是支持的，但Kubeflow是最简单的方式，可能也是未来发展得最快的（基于Kubernetes和容器看起来又轻量级又强大）。

第二步：看一下视频讲解，从overview的角度了解TFX

1.（可选）简单了解TFX的起源，它要解决的问题，以及应用举例。

2.1 （浅显）How to build an ML pipeline with TFX

2.2 （稍专业） TFX是什么东东？

第三步： 趁热打铁，动手创建第一个极简的tfx.dsl.Pipeline

1.先阅读 Understanding TFX Pipelines 了解相关术语。

2.在colab打开这个notebook，开始动手感受一下整体流程。

这个colab大致步骤为：

• Set Up：
  • ……
  • Set up variables
  • Prepare data
• Create a pipeline：
  • Model training code:
  • Pipeline definition:

一个极简的pipeline:tfx.dsl.Pipeline是下面这样子的，只包含3个components：

• 把数据接入到pipeline的 example_gen: tfx.components.CsvExampleGen ；
• trainer:tfx.components.Trainer ；
• pusher:tfx.components.Pusher 。

pipeline的关键代码是这样的：

def _create_pipeline(pipeline_name: str, pipeline_root: str, data_root: str,
                     module_file: str, serving_model_dir: str,
                     metadata_path: str) -> tfx.dsl.Pipeline:
  """Creates a three component penguin pipeline with TFX."""
  # Brings data into the pipeline.
  example_gen = tfx.components.CsvExampleGen(input_base=data_root)

  # Uses user-provided Python function that trains a model.
  trainer = tfx.components.Trainer(
      module_file=module_file,
      examples=example_gen.outputs['examples'],
      train_args=tfx.proto.TrainArgs(num_steps=100),
      eval_args=tfx.proto.EvalArgs(num_steps=5))

  # Pushes the model to a filesystem destination.
  pusher = tfx.components.Pusher(
      model=trainer.outputs['model'],
      push_destination=tfx.proto.PushDestination(
          filesystem=tfx.proto.PushDestination.Filesystem(
              base_directory=serving_model_dir)))

  # Following three components will be included in the pipeline.
  components = [
      example_gen,
      trainer,
      pusher,
  ]

  return tfx.dsl.Pipeline(
      pipeline_name=pipeline_name,
      pipeline_root=pipeline_root,
      metadata_connection_config=tfx.orchestration.metadata
      .sqlite_metadata_connection_config(metadata_path),
      components=components)


# ---

# 调用上面的 `_create_pipeline` 创建 `pipeline:tfx.dsl.Pipeline`， 
# 使用 LocalDagRunner执行就完成了。
pipeline = _create_pipeline(
      pipeline_name=PIPELINE_NAME,
      pipeline_root=PIPELINE_ROOT,
      data_root=DATA_ROOT,
      module_file=_trainer_module_file,
      serving_model_dir=SERVING_MODEL_DIR,
      metadata_path=METADATA_PATH)

# 这里使用了LocalDagRunner（只是在本地机器执行），但是如果要使用其它的orchestrator来跑，
# 除了下方这行代码需要改之外，其它代码几乎不需要改动。
tfx.orchestration.LocalDagRunner().run(pipeline)

其中传递给上面_create_pipeline()的_trainer_module_file的关键代码是这样的：

# ...
# TFX Trainer will call this function.
def run_fn(fn_args: tfx.components.FnArgs):
  """
  所有的参数都在fn_args一串的，不会东一点，西一点到处散乱。
  """
  
  # 对，这里只要提供 tf.keras.Model就行
  model:tf.keras.Model = _build_keras_model()
  model.fit(
      train_dataset,
      steps_per_epoch=fn_args.train_steps,
      validation_data=eval_dataset,
      validation_steps=fn_args.eval_steps)

  # The result of the training should be saved in `fn_args.serving_model_dir`
  # directory.
  model.save(fn_args.serving_model_dir, save_format='tf')

极简总结： CSV数据 ——> example_gen: tfx.components.CsvExampleGen 生成Tensorflow的数据结构 ——> trainer:tfx.components.Trainer ——> pusher:tfx.components.Pusher 。

上面箭头的这些都是 tfx.component，他们构成了一个pipeline:tfx.dsl.Pipeline，pipeline可以被tfx.orchestration.* 下的runner执行，从而完成了数据接入/处理、模型构建、部署整个流程的自动化。

第四步： 加深理解

当然上面第三步只是一个最简的例子， tfx能做的远不止于此：

• 在数据的接入上，上面极简例子用了tfx.components.CsvExampleGen， 除此之外还有多种ExampleGen来接入不同来源的数据，并对训练集/测试集进行精细的处理。
• 上面极简例子中，数据经过example_gen生成样本之后就到了Trainer。现实使用场景中通常都要经过 StatisticsGen ——> SchemaGen ——> Example Validator ——> Transform这几个component， 其中Transform是负责特征工程的，另外几个是负责数据验证的。
• 上面极简例子中，Trainer生成模型之后，就紧接着Pusher了， 现实使用中是要先通过 Evaluators的推导准确率评估和InfraValidator验证推导效率和能耗达标才能Push。
• 除了上面几点之外， TFX Pipeline是支持自定义component的，如果遇到内置component覆盖不到的场景是可以自己编写自定义component来实现的。例如，笔者所在的量化交易领域，AI模型只为自己的推导准确率等指标负责，但是整个产品（策略）追求的是收益率、夏普比率、盈亏比等金融指标，这些指标通常可以通过把AI模型扔到回测系统做历史回测得到。这种情况就可以通过创建自定义component来负责接收上一个component产生的模型，并扔到回测系统执行历史回测来实现。总之TFX Pipeline给了我们许多可能性。

有了这些认识之后，现在可以阅读 https://www.tensorflow.org/tfx/guide 加深理解。

第五步：Kubeflow登场

这里是一个milestone！ 这里是一个milestone！ 这里是一个milestone！ 刚刚的极简例子中我们还没有使用到Kubeflow， 本文的主场从现在这里开始。

上面讲到的这些可以通过 Kubeflow Pipelines 快速入门 体验。

Kubenetes基础知识在TFX MLOps中的意义是？ 在Kubeflow Pipelines使用过程中你会遇到 dockerfile、镜像、volume等kubenetes中常见的概念，因为Kubeflow就是运行在kubenetes集群之上的。 例如，在Kubeflow中，pipeline component的定义是：A pipeline component is a self-contained set of user code, packaged as a Docker image, that performs one step in the pipeline. For example, a component can be responsible for data preprocessing, data transformation, model training, and so on. 而在其它编排器中，pipeline component的定义又不太一样了。 Google 官方的《MLOps (Machine Learning Operations) Fundamentals》 课程也花了很大篇幅来讲Kubenetes基础知识。

Next

学习MLOps不能只关注工具的使用，MLOps认为ML是一个循环渐进的过程，每一次循环，模型性能都应该有所提升。 项目组应该怎么分析把时间花在那环节（工作项）上最能有效的提升模型性能呢？ 以及许多类似的问题，都可以通过吴恩达大师的 《MLOps Specialization 》课程 找到答案或者启发。

或者来自Google官方的的，较难又好像有点缺乏趣味的《MLOps (Machine Learning Operations) Fundamentals》

长按下方二维码识别可在浏览器中阅读：