Memuat data dari Cloud Storage ke BigQuery menggunakan Workflows

Last reviewed 2021-05-12 UTC
Tutorial ini menunjukkan cara menjalankan alur kerja serverless secara andal menggunakan Workflows, Cloud Functions, dan Firestore untuk memuat data mentah, seperti log peristiwa, dari Cloud Storage ke BigQuery. Platform analisis biasanya memiliki alat orkestrasi untuk memuat data secara berkala di BigQuery menggunakan tugas BigQuery, lalu mengubah data tersebut untuk menyediakan metrik bisnis menggunakan pernyataan SQL, termasuk pernyataan bahasa prosedural BigQuery. Tutorial ini ditujukan bagi developer dan arsitek yang ingin membangun pipeline pemrosesan data berbasis peristiwa serverless. Tutorial ini mengasumsikan bahwa Anda sudah terbiasa dengan YAML, SQL, dan Python.

Arsitektur

Diagram berikut menunjukkan arsitektur tingkat tinggi dari pipeline ekstrak, pemuatan, dan transformasi (ELT) tanpa server menggunakan Alur kerja.

Mengekstrak, memuat, dan mentransformasi pipeline.

Pada diagram sebelumnya, pertimbangkan platform retail yang secara berkala mengumpulkan peristiwa penjualan sebagai file dari berbagai toko, lalu menulis file tersebut ke bucket Cloud Storage. Peristiwa tersebut digunakan untuk menyediakan metrik bisnis dengan mengimpor dan memprosesnya di BigQuery. Arsitektur ini menyediakan sistem orkestrasi yang andal dan serverless untuk mengimpor file Anda ke BigQuery, dan dibagi menjadi dua modul berikut:

  • Daftar file: Menyimpan daftar file yang belum diproses yang ditambahkan ke bucket Cloud Storage dalam koleksi Firestore. Modul ini berfungsi melalui Cloud Function yang dipicu oleh peristiwa penyimpanan Object Finalize, yang dihasilkan saat file baru ditambahkan ke bucket Cloud Storage. Nama file ditambahkan ke array files dari koleksi yang bernama new di Firestore.

  • Alur kerja: Menjalankan alur kerja terjadwal. Cloud Scheduler memicu alur kerja yang menjalankan serangkaian langkah sesuai dengan sintaksis berbasis YAML untuk mengatur pemuatan, kemudian mengubah data di BigQuery dengan memanggil Cloud Functions. Langkah-langkah dalam alur kerja memanggil Cloud Functions untuk menjalankan tugas-tugas berikut:

    • Buat dan mulai tugas pemuatan BigQuery.
    • Melakukan polling tentang status tugas pemuatan.
    • Buat dan mulai tugas kueri transformasi.
    • Melakukan polling tentang status tugas transformasi.

Menggunakan transaksi untuk mengelola daftar file baru di Firestore membantu memastikan bahwa tidak ada file yang terlewat saat alur kerja mengimpornya ke BigQuery. Operasi yang terpisah dari alur kerja dibuat idempoten dengan menyimpan metadata dan status tugas di Firestore.

Tujuan

  • Membuat database Firestore.
  • Siapkan pemicu Cloud Function untuk melacak file yang ditambahkan ke bucket Cloud Storage di Firestore.
  • Deploy Cloud Functions untuk menjalankan dan memantau tugas BigQuery.
  • Deploy dan jalankan alur kerja untuk mengotomatiskan prosesnya.

Biaya

Dalam dokumen ini, Anda menggunakan komponen Google Cloud yang dapat ditagih berikut:

Untuk membuat perkiraan biaya berdasarkan proyeksi penggunaan Anda, gunakan kalkulator harga. Pengguna baru Google Cloud mungkin memenuhi syarat untuk mendapatkan uji coba gratis.

Setelah menyelesaikan tugas yang dijelaskan dalam dokumen ini, Anda dapat menghindari penagihan berkelanjutan dengan menghapus resource yang Anda buat. Untuk mengetahui informasi selengkapnya, lihat Pembersihan.

Sebelum memulai

  1. Di konsol Google Cloud, pada halaman pemilih project, pilih atau buat project Google Cloud.

    Buka pemilih project

  2. Pastikan penagihan telah diaktifkan untuk project Google Cloud Anda.

  3. Aktifkan API Cloud Build, Cloud Functions, Identity and Access Management, Resource Manager, and Workflows.

    Mengaktifkan API

  4. Buka halaman Welcome dan catat Project ID untuk digunakan di langkah selanjutnya.

    Buka Halaman sambutan

  5. Di konsol Google Cloud, aktifkan Cloud Shell.

    Aktifkan Cloud Shell

Menyiapkan lingkungan Anda

Untuk menyiapkan lingkungan Anda, buat database Firestore, clone contoh kode dari repo GitHub, buat resource menggunakan Terraform, edit file YAML Workflows, dan persyaratan penginstalan untuk generator file.

  1. Untuk membuat database Firestore, lakukan hal berikut:

    1. Di konsol Google Cloud, buka halaman Firestore.

      Buka Firestore

    2. Klik Pilih mode native.

    3. Di menu Select a location, pilih region tempat Anda ingin menghosting database Firestore. Sebaiknya pilih wilayah yang dekat dengan lokasi fisik Anda.

    4. Klik Buat database.

  2. Pada Cloud Shell, clone repositori sumber:

    cd $HOME && git clone https://github.com/GoogleCloudPlatform/workflows-demos
cd workflows-demos/workflows-bigquery-load

  3. Di Cloud Shell, buat resource berikut menggunakan Terraform:

    terraform init
terraform apply \
    -var project_id=PROJECT_ID \
    -var region=REGION \
    -var zone=ZONE \
    --auto-approve

    Ganti kode berikut:

    • PROJECT_ID: project ID Google Cloud Anda
    • REGION: lokasi geografis Google Cloud tertentu untuk menghosting resource Anda—misalnya, us-central1
    • ZONE: lokasi dalam region untuk menghosting resource Anda—misalnya, us-central1-b

    Anda akan melihat pesan yang mirip dengan yang berikut ini: Apply complete! Resources: 7 added, 0 changed, 1 destroyed.

    Terraform dapat membantu Anda membuat, mengubah, dan mengupgrade infrastruktur dalam skala besar dengan aman dan dapat diprediksi. Resource berikut dibuat dalam project Anda:

    • Akun layanan dengan hak istimewa yang diperlukan untuk memastikan akses yang aman ke resource Anda.
    • Set data BigQuery bernama serverless_elt_dataset dan tabel bernama word_count untuk memuat file yang masuk.
    • Bucket Cloud Storage bernama ${project_id}-ordersbucket untuk file input staging.
    • Lima Cloud Functions berikut:
      • file_add_handler menambahkan nama file yang ditambahkan ke bucket Cloud Storage ke koleksi Firestore.
      • create_job membuat tugas pemuatan BigQuery baru dan mengaitkan file di koleksi Firebase dengan tugas tersebut.
      • create_query membuat tugas kueri BigQuery baru.
      • poll_bigquery_job mendapatkan status tugas BigQuery.
      • run_bigquery_job memulai tugas BigQuery.

  4. Dapatkan URL untuk Cloud Functions create_job, create_query, poll_job, dan run_bigquery_job yang Anda deploy pada langkah sebelumnya.

    gcloud functions describe create_job | grep url
gcloud functions describe poll_bigquery_job | grep url
gcloud functions describe run_bigquery_job | grep url
gcloud functions describe create_query | grep url

    Outputnya mirip dengan hal berikut ini:

    url: https://REGION-PROJECT_ID.cloudfunctions.net/create_job
url: https://REGION-PROJECT_ID.cloudfunctions.net/poll_bigquery_job
url: https://REGION-PROJECT_ID.cloudfunctions.net/run_bigquery_job
url: https://REGION-PROJECT_ID.cloudfunctions.net/create_query

    Catat URL ini karena diperlukan saat Anda men-deploy alur kerja.

Membuat dan men-deploy alur kerja

  1. Di Cloud Shell, buka file sumber untuk alur kerja, workflow.yaml:

    main:
  steps:
    - constants:
        assign:
          - create_job_url: CREATE_JOB_URL
          - poll_job_url: POLL_BIGQUERY_JOB_URL
          - run_job_url: RUN_BIGQUERY_JOB_URL
          - create_query_url: CREATE_QUERY_URL
          - region: BQ_REGION
          - table_name: BQ_DATASET_TABLE_NAME
        next: createJob

    - createJob:
        call: http.get
        args:
          url: ${create_job_url}
          auth:
              type: OIDC
          query:
              region: ${region}
              table_name: ${table_name}
        result: job
        next: setJobId

    - setJobId:
        assign:
          - job_id: ${job.body.job_id}
        next: jobCreateCheck

    - jobCreateCheck:
        switch:
          - condition: ${job_id == Null}
            next: noOpJob
        next: runLoadJob

    - runLoadJob:
        call: runBigQueryJob
        args:
            job_id: ${job_id}
            run_job_url: ${run_job_url}
            poll_job_url: ${poll_job_url}
        result: jobStatus
        next: loadRunCheck

    - loadRunCheck:
        switch:
          - condition: ${jobStatus == 2}
            next: createQueryJob
        next: failedLoadJob

    - createQueryJob:
        call: http.get
        args:
          url: ${create_query_url}
          query:
              qs: "select count(*) from serverless_elt_dataset.word_count"
              region: "US"
          auth:
              type: OIDC
        result: queryjob
        next: setQueryJobId

    - setQueryJobId:
        assign:
          - qid: ${queryjob.body.job_id}
        next: queryCreateCheck

    - queryCreateCheck:
        switch:
          - condition: ${qid == Null}
            next: failedQueryJob
        next: runQueryJob

    - runQueryJob:
        call: runBigQueryJob
        args:
          job_id: ${qid}
          run_job_url: ${run_job_url}
          poll_job_url: ${poll_job_url}
        result: queryJobState
        next: runQueryCheck

    - runQueryCheck:
        switch:
          - condition: ${queryJobState == 2}
            next: allDone
        next: failedQueryJob

    - noOpJob:
        return: "No files to import"
        next: end

    - allDone:
        return: "All done!"
        next: end

    - failedQueryJob:
        return: "Query job failed"
        next: end

    - failedLoadJob:
        return: "Load job failed"
        next: end

runBigQueryJob:
  params: [job_id, run_job_url, poll_job_url]
  steps:
    - startBigQueryJob:
        try:
          call: http.get
          args:
              url: ${run_job_url}
              query:
                job_id: ${job_id}
              auth:
                type: OIDC
              timeout: 600
          result: submitJobState
        retry: ${http.default_retry}
        next: validateSubmit

    - validateSubmit:
        switch:
          - condition: ${submitJobState.body.status == 1}
            next: sleepAndPollLoad
        next: returnState

    - returnState:
        return: ${submitJobState.body.status}

    - sleepAndPollLoad:
        call: sys.sleep
        args:
          seconds: 5
        next: pollJob

    - pollJob:
        try:
          call: http.get
          args:
            url: ${poll_job_url}
            query:
              job_id: ${job_id}
            auth:
              type: OIDC
            timeout: 600
          result: pollJobState
        retry:
          predicate: ${http.default_retry_predicate}
          max_retries: 10
          backoff:
            initial_delay: 1
            max_delay: 60
            multiplier: 2
        next: stateCheck

    - stateCheck:
        switch:
          - condition: ${pollJobState.body.status == 2}
            return: ${pollJobState.body.status}
          - condition: ${pollJobState.body.status == 3}
            return: ${pollJobState.body.status}
        next: sleepAndPollLoad

    Ganti kode berikut:

    • CREATE_JOB_URL: URL fungsi untuk membuat tugas baru
    • POLL_BIGQUERY_JOB_URL: URL fungsi untuk memeriksa status tugas yang sedang berjalan
    • RUN_BIGQUERY_JOB_URL: URL fungsi untuk memulai tugas pemuatan BigQuery
    • CREATE_QUERY_URL: URL fungsi untuk memulai tugas kueri BigQuery
    • BQ_REGION: region BigQuery tempat data disimpan—misalnya, US
    • BQ_DATASET_TABLE_NAME: nama tabel set data BigQuery dalam format PROJECT_ID.serverless_elt_dataset.word_count

  2. Deploy file workflow:

    gcloud workflows deploy WORKFLOW_NAME \
    --location=WORKFLOW_REGION \
    --description='WORKFLOW_DESCRIPTION' \
    --service-account=workflow-runner@PROJECT_ID.iam.gserviceaccount.com \
    --source=workflow.yaml

    Ganti kode berikut:

    • WORKFLOW_NAME: nama unik alur kerja
    • WORKFLOW_REGION: region tempat alur kerja di-deploy—misalnya, us-central1
    • WORKFLOW_DESCRIPTION: deskripsi alur kerja

  3. Buat lingkungan virtual Python 3 dan persyaratan penginstalan untuk generator file:

    sudo apt-get install -y python3-venv
python3 -m venv env
. env/bin/activate
cd generator
pip install -r requirements.txt

Buat file untuk diimpor

Skrip Python gen.py menghasilkan konten acak dalam format Avro. Skemanya sama dengan tabel word_count BigQuery. File Avro ini disalin ke bucket Cloud Storage yang ditentukan.

Di Cloud Shell, buat file:

python gen.py -p PROJECT_ID \
    -o PROJECT_ID-ordersbucket \
    -n RECORDS_PER_FILE \
    -f NUM_FILES \
    -x FILE_PREFIX

Ganti kode berikut:

  • RECORDS_PER_FILE: jumlah data dalam satu file
  • NUM_FILES: jumlah total file yang akan diupload
  • FILE_PREFIX: awalan untuk nama file yang dihasilkan

Melihat entri file di Firestore

Setelah file disalin ke Cloud Storage, Cloud Function handle_new_file akan terpicu. Fungsi ini menambahkan daftar file ke array daftar file dalam dokumen new di koleksi jobs Firestore.

Untuk melihat daftar file, di Konsol Google Cloud, buka halaman Data Firestore.

Buka Data

Daftar file yang ditambahkan ke koleksi.

Memicu alur kerja

Workflows menautkan serangkaian tugas tanpa server dari layanan Google Cloud dan API. Setiap langkah dalam alur kerja ini dijalankan sebagai Cloud Functions dan statusnya disimpan di Firestore. Semua panggilan ke Cloud Functions diautentikasi menggunakan akun layanan dari alur kerja.

Di Cloud Shell, jalankan alur kerja:

gcloud workflows execute WORKFLOW_NAME

Diagram berikut menunjukkan langkah-langkah yang digunakan dalam alur kerja:

Langkah-langkah yang digunakan di alur kerja utama dan sub-alur.

Alur kerja ini dibagi menjadi dua bagian: alur kerja utama dan sub-alur kerja. Alur kerja utama menangani pembuatan tugas dan eksekusi kondisional, sementara sub-alur kerja menjalankan tugas BigQuery. Alur kerja ini melakukan operasi berikut:

  • Cloud Function create_job membuat objek tugas baru, mendapatkan daftar file yang ditambahkan ke Cloud Storage dari dokumen Firestore, dan mengaitkan file tersebut dengan tugas pemuatan. Jika tidak ada file untuk dimuat, fungsi tersebut tidak membuat tugas baru.
  • Cloud Function create_query mengambil kueri yang perlu dijalankan bersama dengan region BigQuery tempat kueri harus dieksekusi. Fungsi ini membuat tugas di Firestore dan menampilkan ID tugas.
  • Cloud Function run_bigquery_job mendapatkan ID tugas yang perlu dijalankan, lalu memanggil BigQuery API untuk mengirim tugas.
  • Daripada menunggu tugas selesai di Cloud Function, Anda dapat mengkueri status tugas secara berkala.
    • Cloud Function poll_bigquery_job memberikan status tugas. Fungsi ini dipanggil berulang kali hingga tugas selesai.
    • Untuk menambahkan penundaan antar-panggilan ke Cloud Function poll_bigquery_job, rutinitas sleep dipanggil dari Workflows.

Melihat status pekerjaan

Anda dapat melihat daftar file dan status tugas.

  1. Di Konsol Google Cloud, buka halaman Data Firestore.

    Buka Data

  2. ID unik (UUID) dibuat untuk setiap tugas. Untuk melihat job_type dan status, klik ID pekerjaan. Setiap pekerjaan mungkin memiliki salah satu jenis dan status berikut:

    • job_type: Jenis tugas yang sedang dijalankan oleh alur kerja dengan salah satu nilai berikut:

      • 0: Memuat data ke BigQuery.
      • 1: Jalankan kueri di BigQuery.

    • status: Status tugas saat ini dengan salah satu nilai berikut:

      • 0: Tugas telah dibuat, tetapi belum dimulai.
      • 1: Tugas sedang berjalan.
      • 2: Tugas berhasil menyelesaikan eksekusinya.
      • 3: Terjadi kesalahan dan pekerjaan tidak berhasil diselesaikan.

    Objek tugas juga berisi atribut metadata seperti region set data BigQuery, nama tabel BigQuery, dan jika berupa tugas kueri, string kueri yang sedang dijalankan.

Daftar file dengan status tugas yang ditandai.

Lihat data di BigQuery

Untuk mengonfirmasi bahwa tugas ELT berhasil, pastikan data muncul dalam tabel.

  1. Di konsol Google Cloud, buka halaman Editor BigQuery.

    Buka Editor

  2. Klik tabel serverless_elt_dataset.word_count.

  3. Klik tab Preview.

    Tab pratinjau yang menampilkan data dalam tabel.

Menjadwalkan alur kerja

Untuk menjalankan alur kerja sesuai jadwal secara berkala, Anda dapat menggunakan Cloud Scheduler.

Pembersihan

Cara termudah untuk menghilangkan penagihan adalah dengan menghapus project Google Cloud yang Anda buat untuk tutorial ini. Atau, Anda dapat menghapus resource satu per satu.

Menghapus resource satu per satu

  1. Di Cloud Shell, hapus semua resource yang dibuat menggunakan Terraform:

    cd $HOME/bigquery-workflows-load
terraform destroy \
-var project_id=PROJECT_ID \
-var region=REGION \
-var zone=ZONE \
--auto-approve

  2. Di Konsol Google Cloud, buka halaman Data Firestore.

    Buka Data

  3. Di samping Jobs, klik Menu, lalu pilih Delete.

    Jalur menu untuk menghapus koleksi.

Menghapus project

  1. Di konsol Google Cloud, buka halaman Manage resource.

    Buka Manage resource

  2. Pada daftar project, pilih project yang ingin Anda hapus, lalu klik Delete.
  3. Pada dialog, ketik project ID, lalu klik Shut down untuk menghapus project.

Langkah selanjutnya