Meluncurkan pipeline Dataflow dengan Cloud Composer

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Catatan: Halaman ini belum direvisi untuk Cloud Composer 3 dan menampilkan konten untuk Cloud Composer 2.

Halaman ini menjelaskan cara menggunakan DataflowTemplateOperator untuk meluncurkan pipeline Dataflow dari Cloud Composer. Pipeline Teks Cloud Storage ke BigQuery adalah pipeline batch yang memungkinkan Anda mengupload file teks yang disimpan di Cloud Storage, mengubahnya menggunakan User Defined Function (UDF) JavaScript yang Anda berikan, dan menampilkan hasilnya ke BigQuery.

Ringkasan

• Sebelum memulai alur kerja, Anda akan membuat entitas berikut:

  • Tabel BigQuery kosong dari set data kosong yang akan menyimpan kolom informasi berikut: location, average_temperature, month, dan, secara opsional, inches_of_rain, is_current, dan latest_measurement.

  • File JSON yang akan menormalisasi data dari file .txt ke dalam format yang benar untuk skema tabel BigQuery. Objek JSON akan memiliki array BigQuery Schema, dengan setiap objek akan berisi nama kolom, jenis input, dan apakah kolom tersebut wajib diisi atau tidak.

  • File .txt input yang akan menyimpan data yang akan diupload secara batch ke tabel BigQuery.

  • Fungsi yang Ditentukan Pengguna yang ditulis dalam JavaScript yang akan mengubah setiap baris file .txt menjadi variabel yang relevan untuk tabel kita.

  • File DAG Airflow yang akan mengarah ke lokasi file ini.

• Selanjutnya, Anda akan mengupload file .txt, file UDF .js, dan file skema .json ke bucket Cloud Storage. Anda juga akan mengupload DAG ke lingkungan Cloud Composer.

• Setelah DAG diupload, Airflow akan menjalankan tugas dari DAG tersebut. Tugas ini akan meluncurkan pipeline Dataflow yang akan menerapkan Fungsi yang Ditetapkan Pengguna ke file .txt dan memformatnya sesuai dengan skema JSON.

• Terakhir, data akan diupload ke tabel BigQuery yang Anda buat sebelumnya.

Sebelum memulai

• Panduan ini memerlukan pemahaman tentang JavaScript untuk menulis User Defined Function.
• Panduan ini mengasumsikan bahwa Anda sudah memiliki lingkungan Cloud Composer. Lihat Membuat lingkungan untuk membuatnya. Anda dapat menggunakan versi Cloud Composer apa pun dengan panduan ini.

• Enable the Cloud Composer, Dataflow, Cloud Storage, BigQuery APIs.

  Enable the APIs

Membuat tabel BigQuery kosong dengan definisi skema

Buat tabel BigQuery dengan definisi skema. Anda akan menggunakan definisi skema ini nanti dalam panduan ini. Tabel BigQuery ini akan menyimpan hasil upload batch.

Penting: Resource BigQuery dan bucket Cloud Storage Anda harus berada di region Compute Engine yang sama.

Untuk membuat tabel kosong dengan definisi skema:

Konsol

1. Di konsol Google Cloud, buka halaman BigQuery:

   Buka BigQuery

2. Di panel navigasi, di bagian Resource, luaskan project Anda.

3. Di panel detail, klik Create dataset.

   

4. Di halaman Create dataset, di bagian Dataset ID, beri nama Set Data Anda average_weather. Biarkan semua kolom lain dalam status defaultnya.

   

5. Klik Create dataset.

6. Kembali ke panel navigasi, di bagian Resource, luaskan project Anda. Kemudian, klik set data average_weather.

7. Di panel detail, klik Buat tabel.

   

8. Di halaman Create table, di bagian Source, pilih Empty table.

9. Di halaman Create table, di bagian Destination:

   • Untuk Dataset name, pilih set data average_weather.

     

   • Di kolom Table name, masukkan nama average_weather.

   • Pastikan Table type disetel ke Native table.

10. Di bagian Schema, masukkan definisi skema. Anda dapat menggunakan salah satu pendekatan berikut:

    • Masukkan informasi skema secara manual dengan mengaktifkan Edit sebagai teks dan memasukkan skema tabel sebagai array JSON. Ketik di kolom berikut:

      [
          {
              "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"
          }
      ]
      

    • Gunakan Tambahkan kolom untuk memasukkan skema secara manual:

      

11. Untuk Partition and cluster settings, gunakan nilai default, No partitioning.

12. Di bagian Advanced options, untuk Encryption, biarkan nilai defaultnya, Google-owned and managed key.

13. Klik Create table.

Catatan: Saat membuat tabel kosong menggunakan Konsol Google Cloud, Anda tidak dapat menambahkan label, deskripsi, atau waktu habis masa berlaku. Anda dapat menambahkan properti opsional ini saat membuat tabel menggunakan CLI atau API. Setelah membuat tabel di Konsol Google Cloud, Anda dapat menambahkan waktu habis masa berlaku, deskripsi, dan label.

bq

Gunakan perintah bq mk untuk membuat set data kosong dan tabel dalam set data ini.

Jalankan perintah berikut untuk membuat set data cuaca global rata-rata:

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

Ganti kode berikut:

• LOCATION: region tempat lingkungan berada.
• PROJECT_ID: Project ID.

Jalankan perintah berikut untuk membuat tabel kosong dalam set data ini dengan definisi skema:

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

Setelah tabel dibuat, Anda dapat memperbarui masa berlaku, deskripsi, dan label tabel. Anda juga dapat mengubah definisi skema.

Python

Simpan kode ini sebagai dataflowtemplateoperator_create_dataset_and_table_helper.py dan perbarui variabel di dalamnya untuk mencerminkan project dan lokasi Anda, lalu jalankan dengan perintah berikut:

python dataflowtemplateoperator_create_dataset_and_table_helper.py

Python

Untuk melakukan autentikasi ke Cloud Composer, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, baca Menyiapkan autentikasi untuk lingkungan pengembangan lokal.

# 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}")

Membuat bucket Cloud Storage

Penting: Resource BigQuery dan bucket Cloud Storage Anda harus berada di region Compute Engine yang sama.

Buat bucket untuk menyimpan semua file yang diperlukan untuk alur kerja. DAG yang Anda buat nanti dalam panduan ini akan mereferensikan file yang Anda upload ke bucket penyimpanan ini. Untuk membuat bucket penyimpanan baru:

Konsol

1. Buka Cloud Storage di konsol Google Cloud.

   Buka Cloud Storage

2. Klik Create Bucket untuk membuka formulir pembuatan bucket.

   1. Masukkan informasi bucket Anda, lalu klik Continue untuk menyelesaikan setiap langkah:

      • Tentukan Nama yang unik secara global untuk bucket Anda. Panduan ini menggunakan bucketName sebagai contoh.

      • Pilih Region untuk jenis lokasi. Selanjutnya, pilih Lokasi tempat data bucket akan disimpan.

      • Pilih Standard sebagai kelas penyimpanan default untuk data Anda.

      • Pilih kontrol akses Seragam untuk mengakses objek Anda.

   2. Klik Done.

gcloud

Gunakan perintah gcloud storage buckets create:

gcloud storage buckets create gs://bucketName/

Ganti kode berikut:

• bucketName: nama bucket yang Anda buat sebelumnya dalam panduan ini.

Contoh kode

C#

Untuk melakukan autentikasi ke Cloud Composer, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, baca Menyiapkan autentikasi untuk lingkungan pengembangan lokal.

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

Untuk melakukan autentikasi ke Cloud Composer, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, baca Menyiapkan autentikasi untuk lingkungan pengembangan lokal.

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

Untuk melakukan autentikasi ke Cloud Composer, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, baca Menyiapkan autentikasi untuk lingkungan pengembangan lokal.

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

Untuk melakukan autentikasi ke Cloud Composer, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, baca Menyiapkan autentikasi untuk lingkungan pengembangan lokal.

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

Untuk melakukan autentikasi ke Cloud Composer, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, baca Menyiapkan autentikasi untuk lingkungan pengembangan lokal.

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

Membuat skema BigQuery berformat JSON untuk tabel output Anda

Buat file skema BigQuery berformat JSON yang cocok dengan tabel output yang Anda buat sebelumnya. Perhatikan bahwa nama, jenis, dan mode kolom harus cocok dengan yang ditentukan sebelumnya dalam skema tabel BigQuery Anda. File ini akan menormalisasi data dari file .txt menjadi format yang kompatibel dengan skema BigQuery Anda. Beri nama file ini 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"
    }]
}

Membuat file JavaScript untuk memformat data

Dalam file ini, Anda akan menentukan UDF (Fungsi yang Ditentukan Pengguna) yang menyediakan logika untuk mengubah baris teks dalam file input. Perhatikan bahwa fungsi ini mengambil setiap baris teks dalam file input Anda sebagai argumennya sendiri, sehingga fungsi akan berjalan satu kali untuk setiap baris file input Anda. Beri nama file ini 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);
}

Membuat file input

File ini akan menyimpan informasi yang ingin Anda upload ke tabel BigQuery. Salin file ini secara lokal dan beri nama 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

Perhatian: Pastikan tidak ada baris kosong di akhir file inputFile.txt untuk menghindari error Dataflow.

Mengupload file ke bucket

Upload file berikut ke bucket Cloud Storage yang Anda buat sebelumnya:

• Skema BigQuery berformat JSON (.json)
• Fungsi yang Ditentukan Pengguna JavaScript (transformCSVtoJSON.js)

• File input teks yang ingin Anda proses (.txt)

Konsol

1. Di Konsol Google Cloud, buka halaman Bucket Cloud Storage.
   

   Buka Buckets

2. Dalam daftar bucket, klik bucket Anda.

3. Di tab Objek untuk bucket, lakukan salah satu hal berikut:

   • Tarik lalu lepas file yang diinginkan dari desktop atau pengelola file ke panel utama di Konsol Google Cloud.

   • Klik tombol Upload File, pilih file yang ingin diupload dalam dialog yang muncul, lalu klik Buka.

gcloud

Jalankan perintah gcloud storage cp:

gcloud storage cp OBJECT_LOCATION gs://bucketName

Ganti kode berikut:

• bucketName: nama bucket yang Anda buat sebelumnya dalam panduan ini.
• OBJECT_LOCATION: jalur lokal ke objek Anda. Contoh, Desktop/transformCSVtoJSON.js.

Contoh kode

Python

Untuk melakukan autentikasi ke Cloud Composer, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, baca Menyiapkan autentikasi untuk lingkungan pengembangan lokal.

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

Untuk melakukan autentikasi ke Cloud Composer, siapkan Kredensial Default Aplikasi. Untuk mengetahui informasi selengkapnya, baca Menyiapkan autentikasi untuk lingkungan pengembangan lokal.

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

Mengonfigurasi DataflowTemplateOperator

Sebelum menjalankan DAG, tetapkan variabel Airflow berikut.

Variabel Airflow	Nilai
project_id	Project ID
gce_zone	Zona Compute Engine tempat cluster Dataflow harus dibuat
bucket_path	Lokasi bucket Cloud Storage yang Anda buat sebelumnya

Sekarang Anda akan mereferensikan file yang dibuat sebelumnya untuk membuat DAG yang memulai alur kerja Dataflow. Salin DAG ini dan simpan secara lokal sebagai 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/",
        },
    )

Mengupload DAG ke Cloud Storage

Upload DAG ke folder /dags di bucket lingkungan Anda. Setelah upload berhasil diselesaikan, Anda dapat melihatnya dengan mengklik link Folder DAG di halaman Lingkungan Cloud Composer.

Melihat status tugas

1. Buka antarmuka web Airflow.
2. Di halaman DAG, klik nama DAG (seperti composerDataflowDAG).
3. Di halaman Detail DAG, klik Tampilan Grafik.

4. Periksa status:

   • Failed: Tugas memiliki kotak merah di sekelilingnya. Anda juga dapat menahan kursor di atas tugas dan mencari Status: Gagal.

   • Success: Tugas memiliki kotak hijau di sekelilingnya. Anda juga dapat mengarahkan kursor ke tugas dan memeriksa Status: Berhasil.

Setelah beberapa menit, Anda dapat memeriksa hasilnya di Dataflow dan BigQuery.

Melihat tugas Anda di Dataflow

1. Di konsol Google Cloud, buka halaman Dataflow.

   Buka Dataflow

2. Tugas Anda diberi nama dataflow_operator_transform_csv_to_bq dengan ID unik yang ditambahkan di akhir nama dengan tanda hubung, seperti ini:

   

3. Klik nama untuk melihat detail tugas.

   

Melihat hasil di BigQuery

1. Di Konsol Google Cloud, buka halaman BigQuery.

   Buka BigQuery

2. Anda dapat mengirimkan kueri menggunakan SQL standar. Gunakan kueri berikut untuk melihat baris yang ditambahkan ke tabel Anda:

   SELECT * FROM projectId.average_weather.average_weather