Memuat data dari Cloud Storage ke BigQuery menggunakan Alur Kerja

Last reviewed 2021-05-12 UTC

Tutorial ini menunjukkan cara menjalankan alur kerja serverless dengan andal menggunakan Workflows, Cloud Run 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 untuk memberikan 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 memahami YAML, SQL, dan Python.

Arsitektur

Diagram berikut menunjukkan arsitektur tingkat tinggi pipeline ekstrak, muat, dan ubah (ELT) serverless menggunakan Workflows.

Dalam 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 ini digunakan untuk memberikan metrik bisnis dengan mengimpor dan memproses di BigQuery. Arsitektur ini menyediakan sistem orkestrasi serverless dan andal untuk mengimpor file ke BigQuery, dan dibagi menjadi dua modul berikut:

• Daftar file: Mempertahankan daftar file yang belum diproses dan ditambahkan ke bucket Cloud Storage dalam koleksi Firestore. Modul ini berfungsi melalui Cloud Run Function yang dipicu oleh peristiwa penyimpanan Object Finalize yang dibuat saat file baru ditambahkan ke bucket Cloud Storage. Nama file ditambahkan ke array files dari koleksi 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, lalu mengubah data di BigQuery dengan memanggil Cloud Run Functions. Langkah-langkah dalam alur kerja memanggil Cloud Run Functions untuk menjalankan tugas berikut:

  • Membuat dan memulai tugas pemuatan BigQuery.
  • Melakukan polling status tugas pemuatan.
  • Membuat dan memulai tugas kueri transformasi.
  • Melakukan polling status tugas transformasi.

Penggunaan transaksi untuk mempertahankan daftar file baru di Firestore membantu memastikan tidak ada file yang terlewat saat alur kerja mengimpornya ke BigQuery. Jalankan alur kerja secara terpisah dan buat idempotent dengan menyimpan metadata dan status tugas di Firestore.

Tujuan

• Membuat database Firestore.
• Menyiapkan pemicu Cloud Run Function untuk melacak file yang ditambahkan ke bucket Cloud Storage di Firestore.
• Men-deploy Cloud Run Functions untuk menjalankan dan memantau tugas BigQuery.
• Men-deploy dan menjalankan alur kerja untuk mengotomatiskan proses.

Biaya

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

• BigQuery
• Cloud Run functions
• Cloud Storage
• Firestore
• Workflows

Untuk membuat perkiraan biaya berdasarkan proyeksi penggunaan Anda, gunakan kalkulator harga.

Pengguna Google Cloud baru 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, baca bagian Pembersihan.

Sebelum memulai

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

   Peran yang diperlukan untuk memilih atau membuat project

   • Memilih project: Memilih project tidak memerlukan peran IAM tertentu Anda dapat memilih project mana pun yang telah diberi peran.
   • Membuat project: Untuk membuat project, Anda memerlukan peran Project Creator (roles/resourcemanager.projectCreator), yang berisi izin resourcemanager.projects.create. Pelajari cara memberikan peran.
   Catatan: Jika Anda tidak berencana menyimpan resource yang dibuat dalam prosedur ini, buat project, bukan memilih project yang sudah ada. Setelah menyelesaikan langkah ini, Anda dapat menghapus project, yang menghapus semua resource yang terkait dengan project ini.

   Buka pemilih project

2. Pastikan penagihan diaktifkan untuk Google Cloud project Anda.

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

   Peran yang diperlukan untuk mengaktifkan API

   Untuk mengaktifkan API, Anda memerlukan izin serviceusage.services.enable. Jika Anda membuat project, kemungkinan Anda sudah memiliki izin ini melalui peran Pemilik (roles/owner). Jika tidak, Anda bisa mendapatkan izin ini melalui peran Service Usage Admin (roles/serviceusage.serviceUsageAdmin). Pelajari cara memberikan peran.

   Aktifkan API

4. Buka halaman Welcome dan catat Project ID untuk digunakan pada langkah berikutnya.

   Buka halaman sambutan

5. Di Google Cloud konsol, 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 instal persyaratan untuk generator file.

1. Untuk membuat database Firestore, lakukan hal berikut:

   1. Di Google Cloud konsol, buka halaman Firestore.

      Buka Firestore

   2. Klik Select native mode.

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

   4. Klik Create 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: ID proyek 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 berikut ini: Apply complete! Resources: 7 added, 0 changed, 1 destroyed.

   Terraform dapat membantu Anda membuat, mengubah, dan mengupgrade infrastruktur dalam skala besar secara aman dan terprediksi. Resource berikut dibuat di 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 masuk.
   • Bucket Cloud Storage bernama ${project_id}-ordersbucket untuk file input staging.
   • Lima Cloud Run 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 dalam 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 Run 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 melakukan polling 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 instal persyaratan untuk generator file:

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

Membuat file untuk diimpor

Skrip Python gen.py membuat 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 rekaman dalam satu file
• NUM_FILES: jumlah total file yang akan diupload
• FILE_PREFIX: awalan untuk nama file yang dibuat

Melihat entri file di Firestore

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

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

Buka Data

Memicu alur kerja

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

Di Cloud Shell, jalankan alur kerja:

gcloud workflows execute WORKFLOW_NAME

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

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

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

Melihat status tugas

Anda dapat melihat daftar file dan status tugas.

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

   Buka Data

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

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

     • 0: Memuat data ke BigQuery.
     • 1: Menjalankan 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 error dan tugas tidak berhasil diselesaikan.

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

Melihat data di BigQuery

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

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

   Buka Editor

2. Klik tabel serverless_elt_dataset.word_count.

3. Klik tab Preview.

   

Menjadwalkan alur kerja

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

Pembersihan

Cara termudah untuk meniadakan tagihan adalah dengan menghapus Google Cloud project 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 Google Cloud konsol, buka halaman Data Firestore.

   Buka Data

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

   

Menghapus project

Perhatian: Menghapus project memiliki efek berikut:
• Semua hal dalam project akan dihapus. Jika menggunakan project yang sudah ada untuk tugas dalam dokumen ini, saat Anda menghapusnya, pekerjaan lain yang telah Anda lakukan dalam project tersebut juga akan terhapus.
• Project ID kustom hilang. Saat membuat project ini, Anda mungkin telah membuat project ID kustom yang ingin digunakan di masa mendatang. Untuk mempertahankan URL yang menggunakan project ID, seperti URL appspot.com , hapus resource yang dipilih di dalam project, bukan menghapus seluruh project.
1. Di Google Cloud konsol, buka halaman Manage resources.

   Buka Kelola resource

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

Langkah berikutnya

• Untuk mempelajari BigQuery lebih lanjut, lihat dokumentasi BigQuery.
• Pelajari cara me mbuat pipeline machine learning kustom serverless.
• Untuk mengetahui lebih banyak tentang arsitektur referensi, diagram, dan praktik terbaik lainnya, jelajahi Pusat Arsitektur Cloud.