Template Cloud Storage Text to BigQuery (Stream)

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 sediakan, dan menambahkan hasilnya ke BigQuery.

Pipeline berjalan tanpa batas waktu dan harus dihentikan secara manual melalui cancel, bukan drain, karena penggunaan transformasi Watch-nya, yang merupakan DoFn yang dapat dibagi yang tidak mendukung pengosongan.

Persyaratan pipeline

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

    Pastikan ada array JSON tingkat atas yang berjudul fields dan bahwa kontennya 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 tersebut, 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 Deskripsi
javascriptTextTransformGcsPath 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.
JSONPath Lokasi Cloud Storage file skema BigQuery Anda, dideskripsikan sebagai JSON. Misalnya: gs://path/to/my/schema.json.
javascriptTextTransformFunctionName Nama fungsi yang ditentukan pengguna (UDF) JavaScript yang ingin Anda gunakan. Misalnya, jika kode fungsi JavaScript Anda adalah myTransform(inJson) { /*...do stuff...*/ }, nama fungsi adalah myTransform. Untuk contoh UDF JavaScript, lihat Contoh UDF.
javascriptTextTransformReloadIntervalMinutes Opsional: Menentukan seberapa sering harus memuat ulang UDF, dalam hitungan menit. Jika nilainya lebih besar dari 0, Dataflow secara berkala akan memeriksa file UDF di Cloud Storage, dan memuat ulang UDF jika file diubah. Parameter ini memungkinkan Anda memperbarui UDF saat pipeline berjalan, tanpa perlu memulai ulang tugas. Jika nilainya adalah 0, pemuatan ulang UDF akan dinonaktifkan. Nilai defaultnya adalah 0.
outputTable Tabel BigQuery yang sepenuhnya memenuhi syarat. Contoh: my-project:dataset.table
inputFilePattern Lokasi Cloud Storage teks yang ingin Anda proses. Misalnya: gs://my-bucket/my-files/text.txt.
bigQueryLoadingTemporaryDirectory Direktori sementara untuk proses pemuatan BigQuery. Contoh: gs://my-bucket/my-files/temp_dir
outputDeadletterTable Tabel untuk pesan yang gagal mencapai tabel output. Misalnya: my-project:dataset.my-unprocessed-table. Jika tidak ada, akan dibuat pada saat eksekusi pipeline. Jika tidak ditentukan, <outputTableSpec>_error_records akan digunakan.

Fungsi yang ditentukan pengguna

Template ini memerlukan UDF yang mengurai file input, seperti yang dijelaskan dalam Persyaratan pipeline. Template tersebut memanggil UDF untuk setiap baris teks di 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
  3. Di kolom Job name, masukkan nama pekerjaan yang unik.
  4. Opsional: Untuk Endpoint regional, pilih nilai dari menu drop-down. Region default-nya adalah us-central1.

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

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

gcloud

Di shell atau terminal Anda, 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 pekerjaan 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 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 fungsi 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 Anda 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 HTTP POST. Untuk 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 pekerjaan 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 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 fungsi 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 Anda untuk pesan yang belum diproses
  • PATH_TO_TEMP_DIR_ON_GCS: jalur Cloud Storage ke direktori sementara

Langkah selanjutnya