Meluncurkan pipeline Dataflow dengan Cloud Composer

Cloud Composer 3 | Cloud Composer 2 | Cloud Composer 1

Halaman ini menjelaskan cara menggunakan DataflowTemplateOperator untuk meluncurkan pipeline Dataflow dari Cloud Composer. Pipeline Cloud Storage Text to BigQuery adalah pipeline batch yang memungkinkan Anda mengupload file teks yang tersimpan di Cloud Storage, mengubahnya menggunakan User Defined Function (UDF) JavaScript yang Anda berikan, dan memberikan 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 input .txt yang akan menyimpan data yang akan diupload secara batch ke tabel BigQuery.

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

  • File DAG Airflow yang akan menunjuk 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 User-Defined Function 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

• Pastikan Anda memiliki izin berikut:

  • Peran Cloud Composer: membuat lingkungan (jika Anda belum memilikinya), mengelola objek di bucket lingkungan, Menjalankan DAG, dan mengakses UI Airflow.
  • Peran Cloud Storage: Membuat bucket dan mengelola objek di dalamnya.
  • Peran BigQuery: Membuat set data dan tabel, mengubah data dalam tabel, mengubah skema dan metadata tabel.
  • Peran Dataflow: Melihat tugas Dataflow.

• Pastikan akun layanan lingkungan Anda memiliki izin untuk membuat tugas Dataflow, mengakses bucket Cloud Storage, dan membaca serta memperbarui data untuk tabel di BigQuery.

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 Buat set data.

   

4. Di halaman Buat set data, di bagian ID set data, 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 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 Opsi lanjutan, untuk Enkripsi, 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 Google Cloud konsol, 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, lihat 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 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 Buat 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 Location tempat data bucket akan disimpan.

      • Pilih Standard sebagai kelas penyimpanan default untuk data Anda.

      • Pilih kontrol akses Seragam untuk mengakses objek Anda.

   2. Klik Selesai.

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, lihat 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, lihat 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, lihat 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, lihat 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, lihat 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

Buat 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 Anda ke dalam 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 Anda

Dalam file ini, Anda akan menentukan UDF (User Defined Function) yang menyediakan logika untuk mengubah baris teks dalam file input. Perhatikan bahwa fungsi ini menggunakan setiap baris teks dalam file input 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);
}

Buat file input Anda

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 Anda

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 Google Cloud konsol, 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, lihat 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, lihat 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

Konfigurasi DataflowTemplateOperator

Sebelum menjalankan DAG, tetapkan variabel Airflow berikut.

Variabel Airflow	Nilai
project_id	Project ID. Contoh: example-project.
gce_zone	Zona Compute Engine tempat cluster Dataflow harus dibuat. Contoh: us-central1-a. Untuk mengetahui informasi selengkapnya tentang zona yang valid, lihat Region dan zona.
bucket_path	Lokasi bucket Cloud Storage yang Anda buat sebelumnya. Contoh: gs://example-bucket.

Sekarang Anda akan mereferensikan file yang Anda buat sebelumnya untuk membuat DAG yang memulai alur kerja Dataflow. Salin DAG ini dan simpan secara lokal sebagai composer-dataflow-dag.py.

Airflow 2

"""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/",
        },
    )

Aliran udara 1

"""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.
  created.
Learn more about the difference between the two here:
https://cloud.google.com/compute/docs/regions-zones
* 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.contrib.operators.dataflow_operator import DataflowTemplateOperator
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 = DataflowTemplateOperator(
        # 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 Anda 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 Graph View.

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 Google Cloud konsol, 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
   

Langkah berikutnya

• Mengelola DAG
• Mengakses UI Airflow