Cloud Composer 1 处于维护后模式。Google 无法发布 Cloud Composer 1 的进一步更新，包括新版 Airflow、问题修复和安全更新。我们建议您规划迁移到 Cloud Composer 2。

此页面由 Cloud Translation API 翻译。

使用 Cloud Composer 启动 Dataflow 流水线

Cloud Composer 1 |Cloud Composer 2 |Cloud Composer 3

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

系统会上传用户定义的函数、输入文件和 JSON 架构
Cloud Storage 存储桶引用这些文件的 DAG 将启动 Dataflow 批处理流水线，该流水线会将用户定义的函数和 JSON 架构文件应用于输入文件。之后，此内容会上传到 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。
Enable the Cloud Composer, Dataflow, Cloud Storage, BigQuery APIs.
Enable the APIs

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

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

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

控制台

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

转到 BigQuery
在导航面板的资源部分中，展开您的项目。
在详细信息面板中，点击创建数据集。
在“创建数据集”页面的数据集 ID 部分中，为数据集 average_weather。保留其他所有字段的默认状态。
点击创建数据集。
返回导航面板，在资源部分中，展开您的项目。然后，点击 average_weather 数据集。
在详细信息面板中，点击创建表。
在创建表页面的来源部分，选择空表。
在创建表页面的目标部分，执行以下操作：
- 对于数据集名称，选择 average_weather 数据集。
- 在表名称字段中，输入名称 average_weather。
- 确认表类型设置为原生表。

在 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"
    }
]

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

对于分区和聚簇设置，保留默认值值：No partitioning。
对于高级选项部分的加密，保留默认值 Google-managed key。
点击创建表。

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 存储桶

创建一个存储桶来存放工作流所需的所有文件。您在本指南稍后创建的 DAG 将引用您上传到此存储桶的文件。要创建新的存储分区，请执行以下操作：

控制台

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

转到 Cloud Storage
点击创建存储分区，打开存储分区创建表单。
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

将文件上传到存储桶

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

JSON 格式的 BigQuery 架构 (.json)
JavaScript 用户定义的函数 (transformCSVtoJSON.js)
您要处理的文本的输入文件 (.txt)

控制台

在 Google Cloud 控制台中，进入 Cloud Storage 存储桶页面。
进入“存储桶”
在存储桶列表中，点击您的存储桶。
在存储桶的对象标签页中，执行以下操作之一：
- 将所需文件从桌面或文件管理器拖放到 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 文件夹链接来查看上传的 DAG。

您环境中的 DAGs 文件夹包含您的 DAG

查看任务的状态

转到 Airflow 网页界面。
在 DAG 页面上，点击 DAG 名称（例如 composerDataflowDAG）。
在 DAG 详细信息页面上，点击 Graph View。
检查状态：
- Failed：任务用红色方框圈住。您还可以将指针悬停在任务上，然后查看 State: Failed。
- Success：任务周围有一个绿色框。您也可以将指针悬停在任务上 状态：成功。

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

在 Dataflow 中查看作业

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

转到 Dataflow
您的作业名为“dataflow_operator_transform_csv_to_bq”，具有唯一 ID 附加一个连字符，例如：
点击名称可查看作业详情。

在 BigQuery 中查看结果

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

转到 BigQuery
您可以使用标准 SQL 提交查询。使用以下查询可查看已添加到表中的行：
```
SELECT * FROM projectId.average_weather.average_weather
```