Template Teks Cloud Storage ke BigQuery (Streaming)

Pipeline Teks Cloud Storage ke BigQuery adalah pipeline streaming yang melakukan streaming file teks yang disimpan di Cloud Storage, mengubahnya menggunakan fungsi yang ditentukan pengguna (UDF) JavaScript yang Anda berikan, dan menambahkan hasilnya ke BigQuery.

Pipeline berjalan tanpa batas waktu dan perlu dihentikan secara manual melalui cancel, bukan drain, karena penggunaan transformasi Watch, yang merupakan DoFn yang dapat dipisah dan tidak mendukung pemindahan.

Persyaratan pipeline

  • Buat file JSON yang mendeskripsikan skema tabel output Anda di BigQuery.

    Pastikan terdapat array JSON level teratas yang berjudul fields dan isinya mengikuti pola {"name": "COLUMN_NAME", "type": "DATA_TYPE"}. Contoh: 

    {
  "fields": [
    {
      "name": "name",
      "type": "STRING"
    },
    {
      "name": "age",
      "type": "INTEGER"
    }
  ]
}
  • Buat file JavaScript (.js) dengan fungsi UDF yang menyediakan logika untuk mengubah baris teks. Fungsi Anda harus menampilkan string JSON.

    Contoh berikut memisahkan setiap baris file CSV, membuat objek JSON dengan nilai, dan menampilkan string JSON:

    function process(inJson) {
  val = inJson.split(",");

  const obj = {
    "name": val[0],
    "age": parseInt(val[1])
  };
  return JSON.stringify(obj);
}

Parameter template

Parameter yang diperlukan

  • inputFilePattern : Jalur gs:// ke teks di Cloud Storage yang ingin Anda proses. (Contoh: gs://bucket-Anda/file-Anda.txt).
  • JSONPath : Jalur gs:// ke file JSON yang menentukan skema BigQuery Anda, yang disimpan di Cloud Storage. (Contoh: gs://your-bucket/your-schema.json).
  • outputTable : Lokasi tabel BigQuery yang akan digunakan untuk menyimpan data yang diproses. Jika Anda menggunakan kembali tabel yang sudah ada, tabel tersebut akan ditimpa. (Contoh: <PROJECT_ID>:<DATASET_NAME>.<TABLE_NAME>).
  • javascriptTextTransformGcsPath : URI Cloud Storage file .js yang menentukan fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan. (Contoh: gs://bucket-anda/transformasi-anda/*.js).
  • javascriptTextTransformFunctionName : Nama fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan. Misalnya, jika kode fungsi JavaScript Anda adalah myTransform(inJson) { /*...do stuff...*/ }, nama fungsinya adalah myTransform. Untuk contoh UDF JavaScript, lihat Contoh UDF (https://github.com/GoogleCloudPlatform/DataflowTemplates#udf-examples) (Contoh: transform_udf1).
  • bigQueryLoadingTemporaryDirectory : Direktori sementara untuk proses pemuatan BigQuery. (Contoh: gs://bucket-anda/file-anda/temp-dir).

Parameter opsional

  • outputDeadletterTable : Tabel untuk pesan yang gagal mencapai tabel output. Jika tidak ada, tabel akan dibuat selama eksekusi pipeline. Jika tidak ditentukan, <outputTableSpec>_error_records akan digunakan. (Contoh: <PROJECT_ID>:<DATASET_NAME>.<TABLE_NAME>).
  • useStorageWriteApiAtLeastOnce : Parameter ini hanya berlaku jika "Gunakan BigQuery Storage Write API" diaktifkan. Jika diaktifkan, semantik minimal satu kali akan digunakan untuk Storage Write API, jika tidak, semantik tepat satu kali akan digunakan. Defaultnya adalah: false.
  • useStorageWriteApi : Jika benar, pipeline akan menggunakan BigQuery Storage Write API (https://cloud.google.com/bigquery/docs/write-api). Nilai defaultnya adalah false. Untuk informasi selengkapnya, lihat Menggunakan Storage Write API (https://beam.apache.org/documentation/io/built-in/google-bigquery/#storage-write-api).
  • numStorageWriteApiStreams : Saat menggunakan Storage Write API, menentukan jumlah aliran tulis. Jika useStorageWriteApi adalah true dan useStorageWriteApiAtLeastOnce adalah false, Anda harus menetapkan parameter ini. Setelan defaultnya adalah: 0.
  • storageWriteApiTriggeringFrequencySec : Saat menggunakan Storage Write API, menentukan frekuensi pemicuan, dalam detik. Jika useStorageWriteApi adalah true dan useStorageWriteApiAtLeastOnce adalah false, Anda harus menetapkan parameter ini.
  • pythonExternalTextTransformGcsPath : Pola jalur Cloud Storage untuk kode Python yang berisi fungsi yang ditentukan pengguna. (Contoh: gs://your-bucket/your-function.py).
  • javascriptTextTransformReloadIntervalMinutes : Menentukan seberapa sering UDF dimuat ulang, dalam hitungan menit. Jika nilainya lebih besar dari 0, Dataflow akan memeriksa file UDF di Cloud Storage secara berkala, dan memuat ulang UDF jika file diubah. Parameter ini memungkinkan Anda mengupdate UDF saat pipeline berjalan, tanpa perlu memulai ulang tugas. Jika nilainya 0, pemuatan ulang UDF akan dinonaktifkan. Nilai defaultnya adalah 0.

Fungsi yang ditentukan pengguna (UDF)

Template ini memerlukan UDF yang mengurai file input, seperti yang dijelaskan dalam Persyaratan pipeline. Template memanggil UDF untuk setiap baris teks dalam setiap file input. Untuk mengetahui informasi selengkapnya tentang cara membuat UDF, lihat Membuat fungsi yang ditentukan pengguna untuk template Dataflow.

Spesifikasi fungsi

UDF memiliki spesifikasi berikut:

  • Input: satu baris teks dari file input.
  • Output: string JSON yang cocok dengan skema tabel tujuan BigQuery.

Menjalankan template

Konsol

  1. Buka halaman Create job from template Dataflow.
    2. Buka Buat tugas dari template
  2. Di kolom Nama tugas, masukkan nama tugas yang unik.
  3. Opsional: Untuk Endpoint regional, pilih nilai dari menu drop-down. Region defaultnya adalah us-central1.

    Untuk mengetahui daftar region tempat Anda dapat menjalankan tugas Dataflow, lihat Lokasi Dataflow.

  4. Dari menu drop-down Dataflow template, pilih the Cloud Storage Text to BigQuery (Stream) template.
  5. Di kolom parameter yang disediakan, masukkan nilai parameter Anda.
  6. Klik Run job.

gcloud

Di shell atau terminal, jalankan template:

gcloud dataflow flex-template run JOB_NAME \
    --template-file-gcs-location gs://dataflow-templates-REGION_NAME/VERSION/flex/Stream_GCS_Text_to_BigQuery_Flex \
    --region REGION_NAME \
    --staging-location STAGING_LOCATION \
    --parameters \
javascriptTextTransformGcsPath=PATH_TO_JAVASCRIPT_UDF_FILE,\
javascriptTextTransformFunctionName=JAVASCRIPT_FUNCTION,\
JSONPath=PATH_TO_BIGQUERY_SCHEMA_JSON,\
inputFilePattern=PATH_TO_TEXT_DATA,\
outputTable=BIGQUERY_TABLE,\
outputDeadletterTable=BIGQUERY_UNPROCESSED_TABLE,\
bigQueryLoadingTemporaryDirectory=PATH_TO_TEMP_DIR_ON_GCS

Ganti kode berikut:

  • JOB_NAME: nama tugas unik pilihan Anda
  • REGION_NAME: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
  • VERSION: versi template yang ingin Anda gunakan

    Anda dapat menggunakan nilai berikut:

  • STAGING_LOCATION: lokasi untuk melakukan staging file lokal (misalnya, gs://your-bucket/staging)
  • JAVASCRIPT_FUNCTION: nama fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan

    Misalnya, jika kode fungsi JavaScript Anda adalah myTransform(inJson) { /*...do stuff...*/ }, nama fungsinya adalah myTransform. Untuk contoh UDF JavaScript, lihat Contoh UDF.

  • PATH_TO_BIGQUERY_SCHEMA_JSON: jalur Cloud Storage ke file JSON yang berisi definisi skema
  • PATH_TO_JAVASCRIPT_UDF_FILE: URI Cloud Storage dari file .js yang menentukan fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan—misalnya, gs://my-bucket/my-udfs/my_file.js
  • PATH_TO_TEXT_DATA: jalur Cloud Storage ke set data teks Anda
  • BIGQUERY_TABLE: nama tabel BigQuery Anda
  • BIGQUERY_UNPROCESSED_TABLE: nama tabel BigQuery untuk pesan yang belum diproses
  • PATH_TO_TEMP_DIR_ON_GCS: jalur Cloud Storage ke direktori sementara

API

Untuk menjalankan template menggunakan REST API, kirim permintaan POST HTTP. Untuk mengetahui informasi selengkapnya tentang API dan cakupan otorisasinya, lihat projects.templates.launch.

POST https://dataflow.googleapis.com/v1b3/projects/PROJECT_ID/locations/LOCATION/flexTemplates:launch
{
   "launch_parameter": {
      "jobName": "JOB_NAME",
      "parameters": {
       "javascriptTextTransformFunctionName": "JAVASCRIPT_FUNCTION",
       "JSONPath": "PATH_TO_BIGQUERY_SCHEMA_JSON",
       "javascriptTextTransformGcsPath": "PATH_TO_JAVASCRIPT_UDF_FILE",
       "inputFilePattern":"PATH_TO_TEXT_DATA",
       "outputTable":"BIGQUERY_TABLE",
       "outputDeadletterTable":"BIGQUERY_UNPROCESSED_TABLE",
       "bigQueryLoadingTemporaryDirectory": "PATH_TO_TEMP_DIR_ON_GCS"
      },
      "containerSpecGcsPath": "gs://dataflow-templates-LOCATION/VERSION/flex/Stream_GCS_Text_to_BigQuery_Flex",
   }
}

Ganti kode berikut:

  • PROJECT_ID: ID project Google Cloud tempat Anda ingin menjalankan tugas Dataflow
  • JOB_NAME: nama tugas unik pilihan Anda
  • LOCATION: region tempat Anda ingin men-deploy tugas Dataflow—misalnya, us-central1
  • VERSION: versi template yang ingin Anda gunakan

    Anda dapat menggunakan nilai berikut:

  • STAGING_LOCATION: lokasi untuk melakukan staging file lokal (misalnya, gs://your-bucket/staging)
  • JAVASCRIPT_FUNCTION: nama fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan

    Misalnya, jika kode fungsi JavaScript Anda adalah myTransform(inJson) { /*...do stuff...*/ }, nama fungsinya adalah myTransform. Untuk contoh UDF JavaScript, lihat Contoh UDF.

  • PATH_TO_BIGQUERY_SCHEMA_JSON: jalur Cloud Storage ke file JSON yang berisi definisi skema
  • PATH_TO_JAVASCRIPT_UDF_FILE: URI Cloud Storage dari file .js yang menentukan fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan—misalnya, gs://my-bucket/my-udfs/my_file.js
  • PATH_TO_TEXT_DATA: jalur Cloud Storage ke set data teks Anda
  • BIGQUERY_TABLE: nama tabel BigQuery Anda
  • BIGQUERY_UNPROCESSED_TABLE: nama tabel BigQuery untuk pesan yang belum diproses
  • PATH_TO_TEMP_DIR_ON_GCS: jalur Cloud Storage ke direktori sementara

Langkah selanjutnya