假设你想要对定期加载到BigQuery表中的数据进行预测建立一个ML模型。你的数据存放在BigQuery中，你希望有一种简单的方式来建立一个ML模型。作为数据分析师，你对建立ML模型了解不多，而且你肯定不了解TensorFlow、Kubeflow Pipelines等框架。

Google Cloud 推出了许多酷炫的工具，帮助数据分析师在他们的工作流程中更好地拥抱和利用机器学习。BQML 正是为了帮助用户构建和使用机器学习模型来实现高级分析而设计的。更具体来说，借助 BQML，数据分析师可以仅通过类似 SQL 的代码来训练、评估和预测。这为数据分析师和分析工程师开启了一个新世界，不仅如此。事实上，对于某些情况，机器学习工程师也开始选择 BQML 作为他们的首选工具。

几乎没有太多问题需要特别考虑，这在某些情况下是必须考虑的，特别是在你想要将ML模型投入生产环境时。

1. 您如何保证您的ML流程（从数据准备到模型训练，再到模型评估和最终预测）始终保持可重复性？
2. 您如何保证您的ML模型在实际预测前已被验证？

为了回答这些问题，ML工程师通常会构建和部署端到端的机器学习流水线。主要有两个平台来实现这个。

1. Kubeflow Pipelines (用于构建和部署机器学习管道的平台)
2. TensorFlow Extended (TFX) (TensorFlow扩展，用于构建和部署生产级机器学习管道的库)

然而，这些平台对数据分析师并不友好。它们需要对Python和容器化技术有很好的理解，这对于数据分析师来说比较困难。这使得数据分析师难以胜任这些平台。数据分析师需要一些类似SQL的工具。这时，Dataform就登场了！

Dataform 是一个 ELT（提取、加载、转换）服务，完全集成在 BigQuery 中的，允许分析团队开发、测试和调度复杂的 SQL 工作流。Dataform 可以被视为在 BigQuery 中通过 SQL 进行数据转换的编排工具。实际上，使用 Dataform，你可以构建可重复和版本化的端到端数据转换管道，该管道基于运行在 BigQuery 上的 SQL 构建。

那么Dataform是如何与机器学习流水线结合的呢？这样的问题可能是每个人在这个时候都会有的疑问。

我们可以利用Dataform的编排功能来创建一个端到端的BQML管道。在这个工作流程中，我们可以按顺序管理模型训练、评估、测试和预测等。

如下图所示，展示了主要想法。

使用 Dataform 的 BQML 流水线编排：主要思想

如图所示，Dataform 运行 ML Pipeline 如下：

1. 数据准备：在这里，Dataform 将 SQL 代码发送到 BigQuery，以执行数据提取、特征选择和特征工程。此步骤将生成用于模型训练和评估的表。
2. 模型训练：在此阶段，Dataform 将 SQL 代码发送到 BigQuery 以激活模型训练。此操作结果是生成一个 ML 模型。
3. 模型评估：在这里，Dataform 将 SQL 代码发送到 BigQuery 以获取相关的 ML 模型指标。根据预设的标准（如准确性、RMSE 等），Dataform 评估模型的质量。如果模型表现可接受，则 ML 流水线继续进行。否则，Dataform 停止其执行。
4. 模型预测：在此阶段，Dataform 将 SQL 代码发送到 BigQuery 对新数据进行推理。结果是一个包含所有预测值的表。

有趣的是，Dataform 担任着编排器的角色。所有的繁重任务都在 BigQuery 上完成。另外，Dataform 还能进行调度，有助于定期重复 ML 管道。

为了更好地理解这项工作是如何进行的，我们将创建一个机器学习流水线，从BigQuery中获取芝加哥出租车行程的公共数据。目标是训练一个深度神经网络，用于预测出租车行程费用。如前所述的那样，机器学习流水线将进行数据准备（包括特征选择和特征工程）、训练模型、评估模型和模型预测。

让我们开始在Dataform里新建一个仓库来存储ML管道代码。提供一个唯一的名字，代码存储的Google Cloud区域，并选择用于运行针对BigQuery的SQL查询的默认服务账户，让Dataform使用它来运行SQL查询。

创建代码库

让我们给Dataform服务账号授予正确的权限。前往IAM & Admin中的IAM，勾选_包含Google提供的角色授予_后，授予BigQuery数据编辑器和BigQuery用户这两个角色，如图所示。

IAM — IAM 服务账户角色（Dataform）

接下来，让我们回到 Dataform，点击新创建的仓库（repo），并创建一个工作区（workspace）。你可以将工作区连接到像 GitHub 这样的 Git 远程仓库。更多详细信息请参阅 文档。

一旦你获得了工作区, 让我们按照以下步骤组织Dataform项目。这是一些指导性建议，而不是严格要求。

Dataform 项目结构

关于项目组织的一些注释如下：

1. 定义/模型：包含所有用于数据准备步骤、模型训练和预测的BQML SQL代码
2. 定义/数据源：包含从数据源提取数据的定义
3. 定义/断言检查：包含用于检查模型性能的模型评估测试
4. workflow_settings.yaml：是项目的主设置文件，包含一些全局设置。

在您按照上述方式整理好Dataform项目之后，我们将在workflow_settings.yaml文件中定义一些全局设置，如下：

    defaultProject: <YOUR GCP PROJECT>  
    defaultLocation: <YOUR BIGQUERY DATASET 位置>  
    defaultDataset: <YOUR 默认 BIGQUERY 数据集 名称，用于 存储 模型 和 表>  
    defaultAssertionDataset: <YOUR 默认 ASSERTIONS 存储 BIGQUERY 数据集 名称>  
    dataformCoreVersion: 3.0.0

这个配置文件已经预先配置好了。您可以设置并选择您喜欢的值。请记得，如果您的数据平台中还没有这些BigQuery数据集，Dataform会为您创建这些数据集。

现在我们准备好继续进行数据准备工作阶段。在这一部分，我们将创建三个包含训练数据集、评估数据集和测试数据集的表。这些表会从芝加哥出租车行程表中获取最新的数据，如下。

1. 训练集将包含从倒数第五个月开始的连续三个月的数据。
2. 评估集将包含从倒数第二个月开始的整整一个月的数据。
3. 测试集将包含最新的一个月。

每个数据集都将会具有以下结构：

表格结构用于训练、评估和测试的数据集的表格结构

我们将选择以下特征来训练模型：

1. dayofweek ：它表示骑行发生的那一周中的星期。范围是从1到7，其中1代表星期日，7代表星期六。
2. hourofday ：它表示骑行发生的具体时刻（指一天中的第几个小时）。
3. trip_miles ：它表示骑行的距离（以英里为单位）。
4. trip_seconds ：它表示骑行的持续时间（以秒为单位）。
5. payment_type ：它告诉我们骑行是如何支付的。
6. company ：它告诉我们提供这项服务的出租车公司名称。

该 标签 结合了行程应付的车费、小费、路桥费和额外费。它是我们的模型预测的目标。

在每个数据集中，我们剔除了行程里程、行程秒数、费用、行程时间戳、支付方式和公司这些特征上有NULL值的记录。此外，我们仅保留了行程里程、行程秒数和费用都大于零的记录。我们筛除了费用超过1500美元的异常值。

在 Dataform 中，我们可以用三个 sqlx 文件来准备这些数据集，这些 sqlx 文件可以包含 JavaScript 代码。实际上，Dataform 会把 sqlx 文件编译成 SQL 代码，并在 BigQuery 上运行。作为编程语言，JavaScript 让分析工程师和数据分析师能写出更高效的 SQL 代码。以下示例展示了如何准备训练数据集的过程。

    -- 文件路径为 /definitions/models/01_data_extractions/chicago_taxi_trips_training.sqlx
    -- 表配置设置：设置材料化策略和自定义数据集名称
    config {  
        type: "table",  
        schema: "df_bqml_chicago_taxi_trips"  
    }  
    -- JavaScript代码块：我们定义了一个特征字段列表。这个列表将在后续查询中使用
    js {  
        const fields_not_null = [  
            'trip_miles',  
            'trip_seconds',  
            'fare',  
            'trip_start_timestamp',  
            'payment_type',  
            'company'  
        ]  
    }  
    -- 获取表中的最新日期
    WITH max_date AS (  
        SELECT DATE(MAX(trip_start_timestamp)) AS latest_date  
        FROM ${ref('taxi_trips')}  
    ),  
    -- 提取最近三个月到两个月的数据集
    data_btw_last3_to_last2_month AS (  
        SELECT *  
        FROM ${ref('taxi_trips')}, max_date  
        WHERE  
    -- 我们选择从5个月前到2个月前的数据。DATE_SUB是BigQuery中的一个函数，允许您从日期中减去一个区间
            DATE(trip_start_timestamp) BETWEEN  
                DATE_SUB(latest_date, INTERVAL 5 MONTH) AND  
                DATE_SUB(latest_date, INTERVAL 2 MONTH)  
    ),  
    -- 应用一些转换和过滤
    final_table_ext AS (  
        SELECT  
            CAST(EXTRACT(DAYOFWEEK FROM trip_start_timestamp) AS INT) AS dayofweek,  
            CAST(EXTRACT(HOUR FROM trip_start_timestamp) AS INT) AS hourofday,  
            trip_miles,  
            trip_seconds,  
            payment_type,  
            company,  
            (fare + tips + tolls + extras) AS label  
        FROM data_btw_last3_to_last2_month  
        WHERE  
            trip_miles > 0  
            AND trip_seconds > 0  
            AND fare BETWEEN 0 AND 1500  
    -- 我们使用JavaScript来提高代码效率。检查当前字段是否非空
            AND ${fields_not_null.map(field => `${field} IS NOT NULL`).join(' AND ')}  
    )  

    SELECT * FROM final_table_ext

我们对评估和测试数据集重复执行相同的操作。当然，我们会相应调整提取区间的设置。评估数据集的时间范围是-2到-1，测试数据集的时间范围是-1到今天。

你可能已经注意到，我们在查询中提到的“taxi_trip”是用来获取数据的。但它是什么，又是从哪里来的呢？在 Dataform 中，我们可以使用 ${ ref( ) } 函数引用数据对象，从而将它们连接起来。特别地，之前的查询引用了一个源表对象数据，该对象提供了定位芝加哥出租车行程表所需的所有设置（在 BigQuery 中，这包括了项目 ID、数据集和表名）。在 Dataform 中，我们可以在 sqlx 文件中声明源数据，如下是一个例子：

配置 {  
    type: "类型",  
    database: "bigquery-public-data",  
    schema: "模式",  
    名称: "taxi_trips"  
}

在 sqlx 文件中声明了源之后，我可以使用 ${ ref('taxi_trips') } 来引用该源。这也会在两个 Dataform 数据对象之间建立隐含的依赖关系。

最终结果如下：项目中新增了四个sqlx文件。

在 Dataform 项目中用于数据准备的 sqlx 文件

此时，我们可以通过运行Dataform执行来在BigQuery中构建这些表格。在这个过程中，Dataform会将我们在SQLX文件中编写的代码编译成适当的SQL，并在BigQuery上执行。结果将会是三个不同的表格，分别包含了提取并过滤的数据。

您可以按照以下截图在Dataform中运行：

数据表操作执行选项

选择执行的动作。然后会出现一个新的右侧窗口。在操作选择菜单中选择您想要执行的数据对象，如下所示：

选择数据表对象的操作选项

结果可以在BigQuery中看到，你会看到三个新的表。如果执行成功，就会这样。

BigQuery 表

很好！现在我们已经得到了数据集，让我们开始训练我们的模型。为此，我们需要在Dataform项目中的definitions/models/02_model_training文件夹下再创建一个sqlx文件。代码如下：

    -- 文件位于 /definitions/models/02_model_training/dnn_reg.sqlx
    -- Dataform 的配置
    config {  
      type: 'operations',  
      hasOutput: true,  
      schema: 'df_bqml_chicago_taxi_trips'  
    }  
    -- 使用 BQML 语法来配置 DNN 模型。  
    CREATE OR REPLACE MODEL ${self()}    
    OPTIONS (  
      MODEL_TYPE = 'DNN_REGRESSOR',  
      ACTIVATION_FN = 'RELU',  
      HIDDEN_UNITS = [64, 64, 64, 64],  
      INPUT_LABEL_COLS = ['label']  
    )  
    AS  
      SELECT  

*  
      FROM  
        ${ref('chicago_taxi_trips_training')}

代码包含了三个主要部分。

1. Dataform 配置：在这里我们指定我们想要的对象类型。对于 BQML，这必须被设置为 operations。然后将 hasOutput 设置为 true 表明此操作将创建一个可以被其他对象通过 ref 函数引用的模型。这很重要，因为当构建评估和预测的 Dataform 对象时，我们需要用到它。最后，我们指定自定义的 BigQuery 数据集（用于存储模型）。
2. ML 模型配置：在这里我们使用 BQML SQL 代码来训练一个回归密集型神经网络。我们指定模型类型（在选项中），每个神经元的激活函数（在此例中我们使用了 ReLu），4 层模型架构（每层包含 64 个神经元，以列表形式配置），以及目标列。
3. 数据集选择：在这里我们使用 SELECT 语句来从训练数据集中选择所有记录。请注意，我们使用了 ${ ref() } 函数来从训练数据集中提取数据。名称是包含提取训练数据集代码的 sqlx 文件的名称。通过这种方式，训练步骤与数据准备步骤之间存在隐含的依赖。

此时你就可以看到你的ML管道是如何逐渐搭建起来的。通过Dataform中的ref()函数，Dataform能够动态地创建一个DAG。这可以在Dataform的编译图部分中看到，如下图所示：

Dataform生成的图表

你可以清楚地看到，你的训练、评估和测试表从声明的源表中提取数据，即我们的芝加哥出租车行程数据表。此外，dnn_reg操作依赖于名为chicago_taxi_trips_training的对象的创建，这个对象当然代表了训练数据集。这真是太酷了！

让我们继续！现在我们的DNN模型已经搭建好了，我们想要评估其性能，然后再进行预测，对不对？所以我们需要运行一些测试来看看我的模型是否符合我们的期望。假设我们只想在平均绝对误差（MAE）低于六美元时才去做预测。因此，我们需要对DNN回归器进行评估，计算MAE，并查看这个值是否低于六。这个过程用BQML和Dataform来做会非常简单。让我们来看看怎么做。

我们将在Dataform中创建一个Assertion。这是一个Dataform提供的工具，可以使用SQL来验证某些条件。我们将在definitions/assertions文件夹中添加一个新的sqlx文件。我们将其命名为evaluate_model.sqlx。代码如下所示：

    -- 文件位于 definitions/assertions/evaluate_model.sqlx
    -- 配置包含类型为断言
    配置为 {  
      type: 'assertion'  
    }  
    -- 我们选择从 ML.EVALUATE BQML SQL 语句返回的所有指标值
    SELECT  

*  
    FROM  
      ML.EVALUATE(  
        MODEL ${ref('dnn_reg')},   
        SELECT * FROM ${ref('chicago_taxi_trips_evaluation')}  
      )  
    WHERE 平均绝对误差 > 6

在之前的代码里，我们需要注意以下几点：

1. 配置指定 Dataform 对象的类型为确认，这是一种特殊对象类型，设计用于确认某些数据条件。在这种情况下，我们确认 MAE 是否大于五美元。如果是，则测试失败，我们将不再继续进行。
2. 我们使用 ML.EVALUATE BQML 代码来进行模型评估。它返回 dnn_reg 模型的所有指标。我们使用 ${ ref() } 函数来引用之前训练的模型和管道初始阶段提取的评估数据。
3. 我们选择所有 MAE 超过六美元的指标。这就是我们确认模型性能的方法。

好的，我们现在来到了我们机器学习管道的最后一步；模型预测阶段。这一步骤只会被执行，当且仅当断言没有失败，也就是说模型的MAE低于六美元。我们需要创建一个新的sqlx文件来使用BQML SQL代码来进行推断。在这里，我们将使用ML.PREDICT，一起使用第三个数据集，也就是测试数据集。现在让我们来看看如何编写这段代码。

    -- 此文件位于路径 /definitions/model/03_model_prediction/predict.sqlx 
    配置 {  
      type: 'table',  
      dependencies: ['evaluate_model'],  
      schema: 'df_bqml_chicago_taxi_trips'  
    }  
    -- 从BQML预测服务中选择特征及预测费用  
    SELECT  
      dayofweek,  
      hourofday,  
      trip_miles,  
      trip_seconds,  
      payment_type,  
      company,  
      ROUND(predicted_label, 2) AS 预测费用  
    FROM  
      ML.PREDICT(  
        MODEL ${ref('dnn_reg')},  
        (  
          SELECT * FROM ${ref('chicago_taxi_trips_testing')}  
        )  
      )

需要注意的是，之前的代码中有一点。

1. 在对象配置中，有一个明确的依赖关系与评估模型的断言相关联。这使得模型预测依赖于断言的正面结果。如果模型评估失败（这意味着MAE大于六美元），预测将不会运行。这确保我们的预测只会在训练模型达到我们的质量标准时才会运行。
2. 我们选择输入到模型中的特征以及预测标签，这是一个由ML.PREDICT BQML SQL语句返回的列。我们决定将其四舍五入到两位，并重命名为predicted_fare。
3. 在ML.PREDICT SQL语句中，我们引用了dnn_reg模型（这产生了一个额外的DAG依赖），并从我们在管道开始阶段提取的测试数据集中选择数据。

数据表DAG的最后部分如下：

这个BQML Dataform Pipeline的最后一步

因此，预测步骤将跟随正确执行后。

1. 将芝加哥出租车行程测试数据集导入到Bigquery的一个表中
2. 训练深度神经网络（DNN）模型
3. 验证测试结果，确保模型的平均绝对误差（MAE）低于五美元

是时候运行整个ML管道执行了。这可以通过点击启动执行，然后打开左侧菜单来完成。打开后，点击“所有操作”按钮，将运行所有Dataform对象，如以下截图所示。

执行所有Datafrom对象的处理

Dataform会根据项目中隐性和显性定义的依赖树自动计算执行顺序。我们可以在“执行”部分的详情里看到这一点，特别是在“编译图”附近，如下图所示。

成功执行的BQML管道（Dataform），

注意：如果你的机器学习管道执行因断言失败而失败，检查模型在BigQuery中的表现（这可以在BigQuery Studio中的模型评估标签页中看到）。相反，如果你有严格的质量要求，你可能需要重新考虑模型类型、架构或特征选择。在BQML中，你可以运行超参数调整作业来找到模型的最佳配置。

好的，现在我们对我们的机器学习流程感到满意了，是时候安排定期运行了。为此，我们需要：

1. 编译机器学习管道代码。我们的机器学习管道需要编译才能在BigQuery上运行。为此，我们必须在Dataform中创建一个Release配置，以便从repo中的一个分支拉取代码进行编译。
2. 将编译后的代码执行。一旦代码被编译，我们就可以安排其在BigQuery上运行。这可以通过设置Dataform的工作流来完成。

可以在 Dataform 仓库页面的 Release & Scheduling 选项卡中如下所示创建发布配置和工作流配置。

chicago_taxi_trips仓库中的发布和调度标签

首先，我们需要创建一个发布版本来指定我们想要从哪个 Git 分支拉取代码。Dataform 本机支持 Git，我们可以在不同的分支中维护代码版本。稳定代码通常被存储在主分支中，而其余代码则存储在其他分支中，例如 dev 或 feat1 等。我们的代码已经在名为 _chicago_taxi_tripworkspace 的分支中开发并提交。如果是这种情况，那么我们将创建一个自定义发布版本，在 Git commitish 文本栏中输入分支名称。否则，我们将创建一个生产版本，使用主分支作为代码源。在以下示例中，我们将创建一个自定义发布版本，并使用工作区名称作为分支的名称。

自定义发布设置

正如我们所见，Dataform每天会在UTC时间7点提取并编译chicago_taxi_trips_workspace中的代码，随后这些代码会在流程配置中使用。

现在我们有了一个发布配置，可以继续创建一个工作流配置了。它的目的是通过定期执行来部署编译代码到BigQuery。在我们的情况下，编译代码将包括整个ML管道及其所有阶段，如数据准备、模型训练、模型评估和模型预测。下图详细展示了工作流配置。

工作流程配置信息

如上图所示，工作流配置需要一个流程配置，该配置说明了需要执行哪些代码。它还需要适当的service account（服务账号）和执行频率（例如，每月一次）。最后，我们将所有操作包含进去，以确保我们管道中的所有机器学习步骤都被包含在内。结果如下：

释放和工作流配置已成功设置

如上图所示，自定义的 Release 和 Workflow 设置已安排好并准备就绪。Dataform 会自动处理这一切，而无需复杂的调度设置。

在这篇文章中，我们讨论了使用BigQuery ML和Dataform首次尝试构建和运行机器学习管道。这两种技术的结合使得数据科学家和分析工程师能够快速构建原型并投入生产BigQuery上的机器学习模型。这些工具的主要好处是降低了他们在BigQuery上采用MLOps的门槛。

这里我们介绍了使用Dataform和BQML的机器学习编排，这为MLOps项目打下了基础。在实际应用中，每个MLOps项目都需要将模型实验、持续训练和持续预测分开进行。确实，在我们提出的解决方案中，模型训练和模型预测位于同一个ML管道中。这样做效率不高，因为每次你想运行预测时，都需要重新训练模型，无论模型性能如何改变。此外，模型训练和预测需要不同的时间表，通常模型训练每个月进行一次，而预测则每天进行一次。比如，你可能会每月重新训练一次模型，但每天运行预测任务。

由于上述原因，我们将在另一篇文章中探讨如何优化我们的MLOps流水线。敬请关注！拜拜