使用 Cloud Composer 启动 Dataflow 流水线

Cloud Composer 1 |Cloud Composer 2 |Cloud Composer 3

注意 ：本页面尚未针对 Cloud Composer 3 进行修订，并且显示 有关 Cloud Composer 2 的内容。

本页介绍了如何使用 DataflowTemplateOperator 启动 Dataflow 流水线 Cloud Composer Cloud Storage Text to BigQuery 流水线 是一种批处理流水线，允许您将存储在 Cloud Storage，使用您提供的 JavaScript 用户定义函数 (UDF) 转换这些函数，并将结果输出到 BigQuery。

概览

• 在开始执行此工作流程之前，您需要创建以下实体：

  • 来自空数据集的空 BigQuery 表，其中包含以下信息列：location、average_temperature、month 以及可选的 inches_of_rain、is_current 和 latest_measurement。

  • 一个 JSON 文件，用于对 .txt 中的数据进行标准化 文件转换为 BigQuery 表的正确格式 架构。JSON 对象将包含一个 BigQuery Schema 数组，其中 每个对象都包含列名称、输入类型以及 而不是必填字段

  • .txt 输入文件，用于保存要批量上传的数据 添加到 BigQuery 表中。

  • 以 JavaScript 编写的用户定义函数，该函数会将 .txt 文件的每一行转换为表的相关变量。

  • Airflow DAG 文件，该文件将指向 文件。

• 接下来，您需要上传 .txt 文件、.js UDF 文件和 .json 架构 复制到 Cloud Storage 存储桶。您还需要将该 DAG 上传到 Cloud Composer 环境。

• 此 DAG 上传后，Airflow 将运行其中的任务。此任务将 启动一个 Dataflow 流水线，该流水线将应用 将用户定义的函数添加到 .txt 文件中，并根据 JSON 架构。

• 最后，将数据上传到 BigQuery 表 创建的新实例

准备工作

• 本指南要求熟悉 JavaScript 来编写代码 用户定义的函数
• 本指南假定您已经拥有 Cloud Composer 环境请参阅创建环境以创建环境。您可以 使用任意版本的 Cloud Composer。

• 启用 Cloud Composer, Dataflow, Cloud Storage, BigQuery API。

  启用 API

创建具有架构定义的空 BigQuery 表

创建具有架构定义的 BigQuery 表。您 在本指南的后面部分，我们将使用此架构定义。这个 BigQuery 表将保存批量上传的结果。

重要提示 ：您的 BigQuery 资源和 Cloud Storage 存储桶必须位于 Compute Engine 区域。

要创建具有架构定义的空表，请执行以下操作：

控制台

1. 在 Google Cloud 控制台中，前往 BigQuery。 页面：

   转到 BigQuery

2. 在导航面板的资源部分中，展开您的项目。

3. 在详细信息面板中，点击创建数据集。

   

4. 在“创建数据集”页面的数据集 ID 部分中，为您的 数据集 average_weather。将其他所有字段保留默认值 状态。

   

5. 点击创建数据集。

6. 返回导航面板，在资源部分展开 自己的项目然后，点击 average_weather 数据集。

7. 在详细信息面板中，点击创建表。

   

8. 在创建表页面的来源部分，选择 空表。

9. 在创建表页面的目标部分，执行以下操作：

   • 对于数据集名称，选择 average_weather 数据集。

     

   • 在表名称字段中，输入名称 average_weather。

   • 确认表类型设置为原生表。

10. 在 Schema 部分中，输入架构定义。您可以使用以下方法之一：

    • 启用以文本形式修改，手动输入架构信息，并 以 JSON 数组格式输入表架构。输入以下内容 字段：

      [
          {
              "name": "location",
              "type": "GEOGRAPHY",
              "mode": "REQUIRED"
          },
          {
              "name": "average_temperature",
              "type": "INTEGER",
              "mode": "REQUIRED"
          },
          {
              "name": "month",
              "type": "STRING",
              "mode": "REQUIRED"
          },
          {
              "name": "inches_of_rain",
              "type": "NUMERIC"
          },
          {
              "name": "is_current",
              "type": "BOOLEAN"
          },
          {
              "name": "latest_measurement",
              "type": "DATE"
          }
      ]
      

    • 使用添加字段手动输入架构：

      

11. 对于分区和聚簇设置，保留默认值 值：No partitioning。

12. 对于高级选项部分的加密，保留 默认值为 Google-managed key。

13. 点击创建表。

。

注意 ：使用 Google Cloud 控制台创建空表时， 您不能添加标签、说明或到期时间。使用 CLI 或 API 创建表时可以添加这些可选属性。在您之后 在 Google Cloud 控制台中创建表，您可以添加到期时间 时间、说明和标签。

bq

使用 bq mk 命令在此创建空数据集和表 数据集。

运行以下命令以创建表示全球平均天气的数据集：

bq --location=LOCATION mk \
    --dataset PROJECT_ID:average_weather

替换以下内容：

• LOCATION：环境所在的区域。
• PROJECT_ID：项目 ID。

运行以下命令在此数据集中创建一个空表： 架构定义：

bq mk --table \
PROJECT_ID:average_weather.average_weather \
location:GEOGRAPHY,average_temperature:INTEGER,month:STRING,inches_of_rain:NUMERIC,is_current:BOOLEAN,latest_measurement:DATE

创建表后，您可以 更新 表的到期时间、说明和标签。也可以修改架构定义。

Python

将此代码另存为 dataflowtemplateoperator_create_dataset_and_table_helper.py 并更新其中的变量以反映您的项目和位置，然后 请使用以下命令运行此测试：

python dataflowtemplateoperator_create_dataset_and_table_helper.py

Python

如需向 Cloud Composer 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。

# Make sure to follow the quickstart setup instructions beforehand.
# See instructions here:
# https://cloud.google.com/bigquery/docs/quickstarts/quickstart-client-libraries

# Before running the sample, be sure to install the bigquery library
# in your local environment by running pip install google.cloud.bigquery

from google.cloud import bigquery

# TODO(developer): Replace with your values
project = "your-project"  # Your GCP Project
location = "US"  # the location where you want your BigQuery data to reside. For more info on possible locations see https://cloud.google.com/bigquery/docs/locations
dataset_name = "average_weather"


def create_dataset_and_table(project, location, dataset_name):
    # Construct a BigQuery client object.
    client = bigquery.Client(project)

    dataset_id = f"{project}.{dataset_name}"

    # Construct a full Dataset object to send to the API.
    dataset = bigquery.Dataset(dataset_id)

    # Set the location to your desired location for the dataset.
    # For more information, see this link:
    # https://cloud.google.com/bigquery/docs/locations
    dataset.location = location

    # Send the dataset to the API for creation.
    # Raises google.api_core.exceptions.Conflict if the Dataset already
    # exists within the project.
    dataset = client.create_dataset(dataset)  # Make an API request.

    print(f"Created dataset {client.project}.{dataset.dataset_id}")

    # Create a table from this dataset.

    table_id = f"{client.project}.{dataset_name}.average_weather"

    schema = [
        bigquery.SchemaField("location", "GEOGRAPHY", mode="REQUIRED"),
        bigquery.SchemaField("average_temperature", "INTEGER", mode="REQUIRED"),
        bigquery.SchemaField("month", "STRING", mode="REQUIRED"),
        bigquery.SchemaField("inches_of_rain", "NUMERIC", mode="NULLABLE"),
        bigquery.SchemaField("is_current", "BOOLEAN", mode="NULLABLE"),
        bigquery.SchemaField("latest_measurement", "DATE", mode="NULLABLE"),
    ]

    table = bigquery.Table(table_id, schema=schema)
    table = client.create_table(table)  # Make an API request.
    print(f"Created table {table.project}.{table.dataset_id}.{table.table_id}")

创建 Cloud Storage 存储桶

重要提示 ：您的 BigQuery 资源和 Cloud Storage 存储桶必须位于 Compute Engine 区域。

创建一个存储桶，以保存该工作流所需的所有文件。您要创建的 DAG 在本指南的后面部分中创建的列表将会引用您上传至此 存储桶。要创建新的存储分区，请执行以下操作：

控制台

1. 在 Google Cloud 控制台中打开 Cloud Storage。

   转到 Cloud Storage

2. 点击创建存储分区，打开存储分区创建表单。

   1. 输入您的存储桶信息，然后点击继续以完成各个步骤：

      • 为您的存储桶指定一个全局唯一的名称。本指南使用的是 以 bucketName 为例。

      • 选择区域作为位置类型。接下来，选择 存储桶数据的存储位置。

      • 选择标准作为数据的默认存储类别。

      • 选择统一访问权限控制以访问您的对象。

   2. 点击完成。

gcloud

使用 gcloud storage buckets create 命令：

gcloud storage buckets create gs://bucketName/

替换以下内容：

• bucketName：您之前在此示例中创建的存储桶的名称 指南。

代码示例

C#

如需向 Cloud Composer 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。

using Google.Apis.Storage.v1.Data;
using Google.Cloud.Storage.V1;
using System;

public class CreateBucketSample
{
    public Bucket CreateBucket(
        string projectId = "your-project-id",
        string bucketName = "your-unique-bucket-name")
    {
        var storage = StorageClient.Create();
        var bucket = storage.CreateBucket(projectId, bucketName);
        Console.WriteLine($"Created {bucketName}.");
        return bucket;
    }
}

Go

如需向 Cloud Composer 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。

import (
	"context"
	"fmt"
	"io"
	"time"

	"cloud.google.com/go/storage"
)

// createBucket creates a new bucket in the project.
func createBucket(w io.Writer, projectID, bucketName string) error {
	// projectID := "my-project-id"
	// bucketName := "bucket-name"
	ctx := context.Background()
	client, err := storage.NewClient(ctx)
	if err != nil {
		return fmt.Errorf("storage.NewClient: %w", err)
	}
	defer client.Close()

	ctx, cancel := context.WithTimeout(ctx, time.Second*30)
	defer cancel()

	bucket := client.Bucket(bucketName)
	if err := bucket.Create(ctx, projectID, nil); err != nil {
		return fmt.Errorf("Bucket(%q).Create: %w", bucketName, err)
	}
	fmt.Fprintf(w, "Bucket %v created\n", bucketName)
	return nil
}

Java

如需向 Cloud Composer 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。

import com.google.cloud.storage.Bucket;
import com.google.cloud.storage.BucketInfo;
import com.google.cloud.storage.Storage;
import com.google.cloud.storage.StorageOptions;

public class CreateBucket {
  public static void createBucket(String projectId, String bucketName) {
    // The ID of your GCP project
    // String projectId = "your-project-id";

    // The ID to give your GCS bucket
    // String bucketName = "your-unique-bucket-name";

    Storage storage = StorageOptions.newBuilder().setProjectId(projectId).build().getService();

    Bucket bucket = storage.create(BucketInfo.newBuilder(bucketName).build());

    System.out.println("Created bucket " + bucket.getName());
  }
}

Python

如需向 Cloud Composer 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。

from google.cloud import storage


def create_bucket(bucket_name):
    """Creates a new bucket."""
    # bucket_name = "your-new-bucket-name"

    storage_client = storage.Client()

    bucket = storage_client.create_bucket(bucket_name)

    print(f"Bucket {bucket.name} created")

Ruby

如需向 Cloud Composer 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。

def create_bucket bucket_name:
  # The ID to give your GCS bucket
  # bucket_name = "your-unique-bucket-name"

  require "google/cloud/storage"

  storage = Google::Cloud::Storage.new
  bucket  = storage.create_bucket bucket_name

  puts "Created bucket: #{bucket.name}"
end

为输出表创建 JSON 格式的 BigQuery 架构

创建与您之前创建的输出表匹配的 JSON 格式的 BigQuery 架构文件。请注意，字段名称、类型和模式 必须与 BigQuery 表之前定义的条目匹配 架构。此文件会将 .txt 文件中的数据标准化为某种格式 与 BigQuery 架构兼容。为此文件命名 jsonSchema.json。

{
    "BigQuery Schema": [
    {
        "name": "location",
        "type": "GEOGRAPHY",
        "mode": "REQUIRED"
    },
    {
        "name": "average_temperature",
        "type": "INTEGER",
        "mode": "REQUIRED"
    },
    {
        "name": "month",
        "type": "STRING",
        "mode": "REQUIRED"
    },
    {
        "name": "inches_of_rain",
        "type": "NUMERIC"
    },
    {
        "name": "is_current",
        "type": "BOOLEAN"
    },
    {
        "name": "latest_measurement",
        "type": "DATE"
    }]
}

创建 JavaScript 文件以设置数据格式

在此文件中，您将定义 UDF（用户定义函数），用于为 转换输入文件中文本行的逻辑。请注意， 函数将输入文件中的每一行文本作为其自己的参数，因此 该函数会针对输入文件的每一行运行一次。为此文件命名 transformCSVtoJSON.js。

function transformCSVtoJSON(line) {
  var values = line.split(',');
  var properties = [
    'location',
    'average_temperature',
    'month',
    'inches_of_rain',
    'is_current',
    'latest_measurement',
  ];
  var weatherInCity = {};

  for (var count = 0; count < values.length; count++) {
    if (values[count] !== 'null') {
      weatherInCity[properties[count]] = values[count];
    }
  }

  return JSON.stringify(weatherInCity);
}

创建输入文件

此文件中会包含您要上传到 BigQuery 表。在本地复制此文件，并将其命名为 inputFile.txt。

POINT(40.7128 74.006),45,'July',null,true,2020-02-16
POINT(41.8781 87.6298),23,'October',13,false,2015-02-13
POINT(48.8566 2.3522),80,'December',null,true,null
POINT(6.5244 3.3792),15,'March',14,true,null

注意： 请确保 inputFile.txt 文件，以避免 Dataflow 错误。

将文件上传到存储桶

将以下文件上传到您创建的 Cloud Storage 存储桶 之前：

• JSON 格式的 BigQuery 架构 (.json)
• JavaScript 用户定义的函数 (transformCSVtoJSON.js)

• 您要处理的文本的输入文件 (.txt)

控制台

1. 在 Google Cloud 控制台中，进入 Cloud Storage 存储桶页面。
   

   进入“存储桶”

2. 在存储桶列表中，点击您的存储桶。

3. 在存储桶的对象标签页中，执行以下操作之一：

   • 将所需文件从桌面或文件管理器拖放到 Google Cloud 控制台的主窗格中。

   • 点击上传文件按钮，选择要 在出现的对话框中点击打开。

gcloud

运行 gcloud storage cp 命令：

gcloud storage cp OBJECT_LOCATION gs://bucketName

替换以下内容：

• bucketName：您之前在 本指南。
• OBJECT_LOCATION：对象的本地路径。例如 Desktop/transformCSVtoJSON.js。

代码示例

Python

如需向 Cloud Composer 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。

from google.cloud import storage


def upload_blob(bucket_name, source_file_name, destination_blob_name):
    """Uploads a file to the bucket."""
    # The ID of your GCS bucket
    # bucket_name = "your-bucket-name"
    # The path to your file to upload
    # source_file_name = "local/path/to/file"
    # The ID of your GCS object
    # destination_blob_name = "storage-object-name"

    storage_client = storage.Client()
    bucket = storage_client.bucket(bucket_name)
    blob = bucket.blob(destination_blob_name)

    # Optional: set a generation-match precondition to avoid potential race conditions
    # and data corruptions. The request to upload is aborted if the object's
    # generation number does not match your precondition. For a destination
    # object that does not yet exist, set the if_generation_match precondition to 0.
    # If the destination object already exists in your bucket, set instead a
    # generation-match precondition using its generation number.
    generation_match_precondition = 0

    blob.upload_from_filename(source_file_name, if_generation_match=generation_match_precondition)

    print(
        f"File {source_file_name} uploaded to {destination_blob_name}."
    )

Ruby

如需向 Cloud Composer 进行身份验证，请设置应用默认凭据。 如需了解详情，请参阅为本地开发环境设置身份验证。

def upload_file bucket_name:, local_file_path:, file_name: nil
  # The ID of your GCS bucket
  # bucket_name = "your-unique-bucket-name"

  # The path to your file to upload
  # local_file_path = "/local/path/to/file.txt"

  # The ID of your GCS object
  # file_name = "your-file-name"

  require "google/cloud/storage"

  storage = Google::Cloud::Storage.new
  bucket  = storage.bucket bucket_name, skip_lookup: true

  file = bucket.create_file local_file_path, file_name

  puts "Uploaded #{local_file_path} as #{file.name} in bucket #{bucket_name}"
end

配置 DataflowTemplateOperator

在运行 DAG 之前，请设置以下 Airflow 变量。

Airflow 变量	值
project_id	项目 ID
gce_zone	Dataflow 集群所在的 Compute Engine 可用区 必须创建
bucket_path	您创建的 Cloud Storage 存储桶的位置 更早

现在，您将引用之前创建的文件来创建启动 Dataflow 工作流的 DAG。复制此 DAG 并将其保存在本地 名称：composer-dataflow-dag.py。

"""Example Airflow DAG that creates a Cloud Dataflow workflow which takes a
text file and adds the rows to a BigQuery table.

This DAG relies on four Airflow variables
https://airflow.apache.org/docs/apache-airflow/stable/concepts/variables.html
* project_id - Google Cloud Project ID to use for the Cloud Dataflow cluster.
* gce_zone - Google Compute Engine zone where Cloud Dataflow cluster should be
  created.
For more info on zones where Dataflow is available see:
https://cloud.google.com/dataflow/docs/resources/locations
* bucket_path - Google Cloud Storage bucket where you've stored the User Defined
Function (.js), the input file (.txt), and the JSON schema (.json).
"""

import datetime

from airflow import models
from airflow.providers.google.cloud.operators.dataflow import (
    DataflowTemplatedJobStartOperator,
)
from airflow.utils.dates import days_ago

bucket_path = "{{var.value.bucket_path}}"
project_id = "{{var.value.project_id}}"
gce_zone = "{{var.value.gce_zone}}"


default_args = {
    # Tell airflow to start one day ago, so that it runs as soon as you upload it
    "start_date": days_ago(1),
    "dataflow_default_options": {
        "project": project_id,
        # Set to your zone
        "zone": gce_zone,
        # This is a subfolder for storing temporary files, like the staged pipeline job.
        "tempLocation": bucket_path + "/tmp/",
    },
}

# Define a DAG (directed acyclic graph) of tasks.
# Any task you create within the context manager is automatically added to the
# DAG object.
with models.DAG(
    # The id you will see in the DAG airflow page
    "composer_dataflow_dag",
    default_args=default_args,
    # The interval with which to schedule the DAG
    schedule_interval=datetime.timedelta(days=1),  # Override to match your needs
) as dag:
    start_template_job = DataflowTemplatedJobStartOperator(
        # The task id of your job
        task_id="dataflow_operator_transform_csv_to_bq",
        # The name of the template that you're using.
        # Below is a list of all the templates you can use.
        # For versions in non-production environments, use the subfolder 'latest'
        # https://cloud.google.com/dataflow/docs/guides/templates/provided-batch#gcstexttobigquery
        template="gs://dataflow-templates/latest/GCS_Text_to_BigQuery",
        # Use the link above to specify the correct parameters for your template.
        parameters={
            "javascriptTextTransformFunctionName": "transformCSVtoJSON",
            "JSONPath": bucket_path + "/jsonSchema.json",
            "javascriptTextTransformGcsPath": bucket_path + "/transformCSVtoJSON.js",
            "inputFilePattern": bucket_path + "/inputFile.txt",
            "outputTable": project_id + ":average_weather.average_weather",
            "bigQueryLoadingTemporaryDirectory": bucket_path + "/tmp/",
        },
    )

将 DAG 上传到 Cloud Storage

将您的 DAG 上传到环境的/dags 存储桶。上传成功完成后，您可以通过 点击 Cloud Composer 上的 DAG 文件夹链接 “环境”页面。

查看任务状态

1. 转到 Airflow 网页界面。
2. 在 DAG 页面上，点击 DAG 名称（例如 composerDataflowDAG）。
3. 在 DAG 详细信息页面上，点击 Graph View。

4. 检查状态：

   • Failed：任务用红色方框圈住。 您也可以将指针悬停在任务上，然后查看是否有 State: Failed。

   • Success：任务被一个绿色框圈起来。 您也可以将指针悬停在任务上 状态：成功。

几分钟后，您可以在 Dataflow 中查看结果 BigQuery。

在 Dataflow 中查看作业

1. 在 Google Cloud 控制台中，转到 Dataflow 页面。

   转到 Dataflow

2. 您的作业名为“dataflow_operator_transform_csv_to_bq”，具有唯一 ID 附加一个连字符，例如：

   

3. 点击名称即可查看 职位信息详情。

   

在 BigQuery 中查看结果

1. 在 Google Cloud 控制台中，转到 BigQuery 页面。

   转到 BigQuery

2. 您可以使用标准 SQL 提交查询。使用以下查询 查看已添加到表格中的行：

   SELECT * FROM projectId.average_weather.average_weather