Membuat kueri berkelanjutan

Dokumen ini menjelaskan cara menjalankan kueri berkelanjutan di BigQuery.

Kueri berkelanjutan BigQuery adalah pernyataan SQL yang berjalan terus-menerus dan memproses data hampir secara real-time. Anda dapat menulis atau mengekspor baris output yang dihasilkan oleh kueri berkelanjutan ke tujuan berikut:

Pilih jenis akun

Anda dapat membuat dan menjalankan tugas kueri berkelanjutan menggunakan akun pengguna, atau Anda dapat membuat tugas kueri berkelanjutan menggunakan akun pengguna, lalu menjalankannya menggunakan akun layanan. Anda harus menggunakan akun layanan untuk menjalankan kueri berkelanjutan yang mengekspor hasil ke topik Pub/Sub.

Saat Anda menggunakan akun pengguna, kueri berkelanjutan akan berjalan hingga dua hari. Saat Anda menggunakan akun layanan, kueri berkelanjutan akan berjalan hingga 150 hari. Untuk mengetahui informasi selengkapnya, lihat Otorisasi.

Izin yang diperlukan

Bagian ini menjelaskan izin yang Anda perlukan untuk membuat dan menjalankan kueri berkelanjutan. Sebagai alternatif untuk peran Identity and Access Management (IAM) yang disebutkan, Anda bisa mendapatkan izin yang diperlukan melalui peran khusus.

Izin saat menggunakan akun pengguna

Bagian ini memberikan informasi tentang peran dan izin yang diperlukan untuk membuat dan menjalankan kueri berkelanjutan menggunakan akun pengguna.

Untuk membuat tugas di BigQuery, akun pengguna harus memiliki izin IAM bigquery.jobs.create. Setiap peran IAM berikut memberikan izin bigquery.jobs.create:

Untuk mengekspor data dari tabel BigQuery, akun pengguna harus memiliki izin IAM bigquery.tables.export. Setiap peran IAM berikut memberikan izin bigquery.tables.export:

Untuk menulis data ke tabel BigQuery, akun pengguna harus memiliki izin IAM bigquery.tables.updateData. Setiap peran IAM berikut memberikan izin bigquery.tables.updateData:

Jika akun pengguna harus mengaktifkan API yang diperlukan untuk kasus penggunaan kueri berkelanjutan, akun pengguna harus memiliki peran Service Usage Admin (roles/serviceusage.serviceUsageAdmin).

Untuk menulis data ke tabel terkelola Apache Iceberg, selain izin BigQuery, akun pengguna juga harus memiliki peran BigQuery Connection User (roles/bigquery.connectionUser) untuk menggunakan koneksi resource Google Cloud yang terkait dengan tabel terkelola Iceberg. Selain itu, akun layanan yang terkait dengan koneksi tersebut harus memiliki izin yang sesuai (misalnya, roles/storage.objectUser dan roles/storage.legacyBucketReader) pada bucket Cloud Storage yang mendasarinya. Untuk mengetahui informasi selengkapnya, lihat Membuat dan menggunakan tabel terkelola Apache Iceberg.

Izin saat menggunakan akun layanan

Bagian ini memberikan informasi tentang peran dan izin yang diperlukan oleh akun pengguna yang membuat kueri berkelanjutan, dan akun layanan yang menjalankan kueri berkelanjutan.

Izin akun pengguna

Untuk membuat tugas di BigQuery, akun pengguna harus memiliki izin IAM bigquery.jobs.create. Setiap peran IAM berikut memberikan izin bigquery.jobs.create:

Untuk mengirimkan tugas yang berjalan menggunakan akun layanan, akun pengguna harus memiliki peran Service Account User (roles/iam.serviceAccountUser). Jika Anda menggunakan akun pengguna yang sama untuk membuat akun layanan, akun pengguna tersebut harus memiliki peran Service Account Admin (roles/iam.serviceAccountAdmin). Untuk mengetahui informasi tentang cara membatasi akses pengguna ke satu akun layanan, bukan ke semua akun layanan dalam sebuah project, lihat Memberikan satu peran.

Jika akun pengguna harus mengaktifkan API yang diperlukan untuk kasus penggunaan kueri berkelanjutan, akun pengguna harus memiliki peran Service Usage Admin (roles/serviceusage.serviceUsageAdmin).

Izin akun layanan

Untuk mengekspor data dari tabel BigQuery, akun layanan harus memiliki izin IAM bigquery.tables.export. Setiap peran IAM berikut memberikan izin bigquery.tables.export:

Untuk menulis data ke tabel BigQuery, akun layanan harus memiliki izin IAM bigquery.tables.updateData. Setiap peran IAM berikut memberikan izin bigquery.tables.updateData:

Untuk menulis data ke tabel terkelola Apache Iceberg, selain izin BigQuery, akun layanan juga harus memiliki peran BigQuery Connection User (roles/bigquery.connectionUser) untuk menggunakan koneksi resource Google Cloud yang terkait dengan tabel terkelola Iceberg. Selain itu, akun layanan yang terkait dengan koneksi tersebut harus memiliki izin yang sesuai (misalnya, roles/storage.objectUser dan roles/storage.legacyBucketReader) pada bucket Cloud Storage yang mendasarinya. Untuk mengetahui informasi selengkapnya, lihat Membuat dan menggunakan tabel terkelola Apache Iceberg.

Sebelum memulai

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

    Peran yang diperlukan untuk memilih atau membuat project

    • Memilih project: Memilih project tidak memerlukan peran IAM tertentu—Anda dapat memilih project apa pun yang perannya telah diberikan kepada Anda.
    • Membuat project: Untuk membuat project, Anda memerlukan peran Project Creator (roles/resourcemanager.projectCreator), yang berisi izin resourcemanager.projects.create. Pelajari cara memberikan peran.

    Buka pemilih project

  2. Pastikan penagihan diaktifkan untuk project Google Cloud Anda.

  3. Aktifkan BigQuery API, jika belum diaktifkan.

    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.

    Mengaktifkan API

Membuat pemesanan

Buat pemesanan edisi Enterprise atau Enterprise Plus, lalu buat penetapan pemesanan dengan jenis tugas CONTINUOUS. Pemesanan ini dapat menggunakan penskalaan otomatis dan pembagian slot tidak ada aktivitas. Ada batasan pemesanan yang berlaku untuk penetapan pemesanan untuk kueri berkelanjutan.

Ekspor ke Pub/Sub

API tambahan, izin IAM, dan resource Google Cloud diperlukan untuk mengekspor data ke Pub/Sub. Untuk mengetahui informasi selengkapnya, lihat Mengekspor ke Pub/Sub.

Memproses mutasi dengan CHANGES

Saat mengekspor data ke Pub/Sub, Anda memiliki opsi untuk menggunakan fungsi histori perubahan CHANGES. Fungsi CHANGES memproses semua baris yang telah berubah dalam tabel sumber, termasuk penambahan dan mutasi.

Menyematkan atribut kustom sebagai metadata dalam pesan Pub/Sub

Anda dapat menggunakan atribut Pub/Sub untuk memberikan informasi tambahan tentang pesan, seperti prioritas, asal, tujuan, atau metadata tambahan. Anda juga dapat menggunakan atribut untuk memfilter pesan pada langganan.

Dalam hasil kueri berkelanjutan, jika kolom diberi nama _ATTRIBUTES, nilainya akan disalin ke atribut pesan Pub/Sub. Kolom yang disediakan dalam _ATTRIBUTES digunakan sebagai kunci atribut.

Kolom _ATTRIBUTES harus berjenis JSON, dalam format ARRAY<STRUCT<STRING, STRING>> atau STRUCT<STRING>.

Sebagai contoh, lihat mengekspor data ke topik Pub/Sub.

Mengekspor ke Bigtable

API, izin IAM, dan resource Google Cloudtambahan diperlukan untuk mengekspor data ke Bigtable. Untuk mengetahui informasi selengkapnya, lihat Mengekspor ke Bigtable.

Mengekspor ke Spanner

API tambahan, izin IAM, dan resource Google Clouddiperlukan untuk mengekspor data ke Spanner. Untuk mengetahui informasi selengkapnya, lihat Mengekspor ke Spanner (ETL terbalik).

Menulis data ke tabel BigQuery atau tabel terkelola Apache Iceberg

Anda dapat menulis data ke tabel BigQuery atau tabel terkelola Iceberg menggunakan pernyataan INSERT.

Saat menulis output kueri berkelanjutan ke tabel terkelola Iceberg:

  • Tabel terkelola Iceberg tujuan harus sudah ada sebelum Anda membuat kueri berkelanjutan. Untuk mengetahui petunjuk cara membuat tabel terkelola Iceberg, lihat Membuat dan menggunakan tabel terkelola Apache Iceberg.
  • Sintaksis kueri sama dengan penulisan ke tabel BigQuery standar. Anda dapat menentukan tabel tujuan dalam klausa INSERT INTO dan tidak perlu menentukan opsi koneksi atau bucket dalam pernyataan SQL.
  • Akun pengguna atau akun layanan yang menjalankan kueri berkelanjutan harus memiliki izin yang diperlukan pada tabel tujuan, koneksi resource Google Cloud , dan bucket Cloud Storage yang mendasarinya. Untuk mengetahui informasi selengkapnya, lihat Izin yang diperlukan.

Sebagai contoh, lihat Menulis data ke tabel BigQuery atau tabel terkelola Apache Iceberg.

Menggunakan fungsi AI

API, izin IAM, dan resource Google Cloud tambahan diperlukan untuk menggunakan fungsi AI yang didukung dalam kueri berkelanjutan. Untuk mengetahui informasi selengkapnya, lihat salah satu topik berikut, berdasarkan kasus penggunaan Anda:

Saat Anda menggunakan fungsi AI dalam kueri berkelanjutan, pertimbangkan apakah output kueri akan tetap berada dalam kuota untuk fungsi tersebut. Jika melebihi kuota, Anda mungkin harus menangani secara terpisah data yang tidak diproses.

Menentukan waktu mulai untuk kueri berkelanjutan

Anda harus menggunakan fungsi APPENDS, atau fungsi CHANGES dalam kasus ekspor ke Pub/Sub, dalam klausa FROM kueri berkelanjutan untuk menentukan data paling awal yang akan diproses. Misalnya, APPENDS(TABLE my_table, start_timestamp).

Argumen start_timestamp menentukan waktu saat kueri berkelanjutan mulai memproses data. Misalnya, APPENDS(TABLE my_table, CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) memberi tahu BigQuery untuk memproses data yang ditambahkan ke tabel my_table paling lambat 10 menit sebelum kueri berkelanjutan dimulai. Data yang kemudian ditambahkan ke my_table akan diproses saat masuk. Tidak ada penundaan yang diberlakukan pada pemrosesan data.

Saat menentukan argumen start_timestamp, nilai harus berada dalam periode perjalanan waktu tabel, yang secara default adalah tujuh hari untuk tabel standar. Menetapkan start_timestamp ke NULL akan ditetapkan secara default ke waktu pembuatan tabel. Penggunaan nilai NULL tidak direkomendasikan karena error akan ditampilkan jika tabel dibuat lebih awal dari periode perjalanan waktu. Kueri yang menggunakan NULL mungkin berhasil untuk tabel yang baru dibuat, tetapi akan gagal nanti jika stempel waktu pembuatan tabel sudah melewati periode tujuh hari.

Jangan berikan argumen end_timestamp ke fungsi APPENDS saat Anda menggunakannya dalam kueri berkelanjutan.

Contoh berikut menunjukkan cara memulai kueri berkelanjutan dari titik waktu tertentu menggunakan fungsi APPENDS, saat membuat kueri tabel BigQuery yang menerima informasi perjalanan taksi streaming:

EXPORT DATA
  OPTIONS (format = 'CLOUD_PUBSUB',
    uri = 'https://pubsub.googleapis.com/projects/myproject/topics/taxi-real-time-rides') AS (
  SELECT
    TO_JSON_STRING(STRUCT(ride_id,
        timestamp,
        latitude,
        longitude)) AS message
  FROM
    APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
      -- Configure the APPENDS TVF start_timestamp to specify when you want to
      -- start processing data using your continuous query.
      -- This example starts processing at 10 minutes before the current time.
      CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE
    ride_status = 'enroute');

Menentukan titik awal sebelum periode perjalanan waktu

Untuk menyertakan data yang berada di luar periode perjalanan waktu tujuh hari, gunakan kueri standar untuk mengisi ulang data hingga titik waktu tertentu, lalu mulai kueri berkelanjutan dari titik tersebut. Stempel waktu "serah terima" harus berada dalam periode tujuh hari agar kueri berkelanjutan dapat melanjutkan dari bagian terakhir yang ditinggalkan kueri standar.

Contoh berikut menunjukkan cara mengisi ulang data lama dari tabel BigQuery yang menerima informasi perjalanan taksi streaming, lalu beralih ke kueri berkelanjutan.

  1. Jalankan kueri standar untuk mengisi ulang data hingga titik waktu tertentu:

    INSERT INTO `myproject.real_time_taxi_streaming.transformed_taxirides`
SELECT
  timestamp,
  meter_reading,
  ride_status,
  passenger_count,
  ST_Distance(
    ST_GeogPoint(pickup_longitude, pickup_latitude),
    ST_GeogPoint(dropoff_longitude, dropoff_latitude)) AS euclidean_trip_distance,
    SAFE_DIVIDE(meter_reading, passenger_count) AS cost_per_passenger
FROM `myproject.real_time_taxi_streaming.taxirides`
  -- Include all data inserted into the table up to this handoff point.
  -- This handoff timestamp must be within the time travel window.
  FOR SYSTEM_TIME AS OF '2025-01-01 00:00:00 UTC'
WHERE
  ride_status = 'dropoff';

  2. Jalankan kueri berkelanjutan dari titik waktu saat kueri berhenti:

    INSERT INTO `myproject.real_time_taxi_streaming.transformed_taxirides`
SELECT
  timestamp,
  meter_reading,
  ride_status,
  passenger_count,
  ST_Distance(
    ST_GeogPoint(pickup_longitude, pickup_latitude),
    ST_GeogPoint(dropoff_longitude, dropoff_latitude)) AS euclidean_trip_distance,
    SAFE_DIVIDE(meter_reading, passenger_count) AS cost_per_passenger
FROM
  APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
    -- Configure the APPENDS TVF start_timestamp to start processing
    -- data right where the batch query left off + 1 microsecond.
    -- This timestamp must be within the time travel window.
    TIMESTAMP '2025-01-01 00:00:00 UTC' + INTERVAL 1 MICROSECOND)
WHERE
  ride_status = 'dropoff';

Menjalankan kueri berkelanjutan menggunakan akun pengguna

Bagian ini menjelaskan cara menjalankan kueri berkelanjutan menggunakan akun pengguna. Setelah kueri berkelanjutan berjalan, Anda dapat menutup konsol Google Cloud , jendela terminal, atau aplikasi tanpa mengganggu eksekusi kueri. Kueri berkelanjutan yang dijalankan oleh akun pengguna berjalan selama maksimum dua hari, lalu berhenti secara otomatis. Untuk terus memproses data baru yang masuk, mulai kueri berkelanjutan baru dan tentukan titik awal. Untuk mengotomatiskan proses ini, lihat mencoba ulang kueri yang gagal.

Ikuti langkah-langkah berikut untuk menjalankan kueri berkelanjutan:

Konsol

  1. Di konsol Google Cloud , buka halaman BigQuery.

    Buka BigQuery

  2. Di editor kueri, klik Edit > Query mode > Kueri berkelanjutan. Untuk mengonfirmasi pilihan ini, klik Konfirmasi.

  3. Opsional: Untuk mengontrol durasi kueri berjalan, klik Edit > Query settings dan tetapkan Job timeout dalam milidetik.

  4. Di editor kueri, ketik pernyataan SQL untuk kueri berkelanjutan. Pernyataan SQL hanya boleh berisi operasi yang didukung.

  5. Klik Run.

bq

  1. Di konsol Google Cloud , aktifkan Cloud Shell.

    Aktifkan Cloud Shell

    Di bagian bawah konsol Google Cloud , sesi Cloud Shell akan dimulai dan menampilkan prompt command line. Cloud Shell adalah lingkungan shell dengan Google Cloud CLI yang sudah terinstal, dan dengan nilai yang sudah ditetapkan untuk project Anda saat ini. Diperlukan waktu beberapa detik untuk melakukan inisialisasi pada sesi.

  2. Di Cloud Shell, jalankan kueri berkelanjutan menggunakan perintah bq query dengan flag --continuous:

    bq query --use_legacy_sql=false --continuous=true
'QUERY'

    Ganti QUERY dengan pernyataan SQL untuk kueri berkelanjutan. Pernyataan SQL hanya boleh berisi operasi yang didukung. Anda dapat mengontrol durasi kueri berjalan menggunakan flag --job_timeout_ms.

API

Jalankan kueri berkelanjutan dengan memanggil metode jobs.insert. Anda harus menetapkan kolom continuous ke true di JobConfigurationQuery dari resource Job yang Anda teruskan. Anda dapat mengontrol berapa lama kueri berjalan dengan menetapkan kolom jobTimeoutMs.

curl --request POST \
  "https://bigquery.googleapis.com/bigquery/v2/projects/PROJECT_ID/jobs" \
  --header "Authorization: Bearer $(gcloud auth print-access-token)" \
  --header "Content-Type: application/json; charset=utf-8" \
  --data '{"configuration":{"query":{"query":"QUERY","useLegacySql":false,"continuous":true}}}' \
  --compressed

Ganti kode berikut:

  • PROJECT_ID: project ID Anda.
  • QUERY: pernyataan SQL untuk kueri berkelanjutan. Pernyataan SQL hanya boleh berisi operasi yang didukung.

Menjalankan kueri berkelanjutan menggunakan akun layanan

Bagian ini menjelaskan cara menjalankan kueri berkelanjutan menggunakan akun layanan. Setelah kueri berkelanjutan berjalan, Anda dapat menutup konsol Google Cloud , jendela terminal, atau aplikasi tanpa mengganggu eksekusi kueri. Kueri berkelanjutan yang dijalankan menggunakan akun layanan dapat berjalan hingga 150 hari, lalu berhenti secara otomatis. Untuk terus memproses data baru yang masuk, mulai kueri berkelanjutan baru dan tentukan titik awal. Untuk mengotomatiskan proses ini, lihat mencoba ulang kueri yang gagal.

Jika Anda menggunakan akun layanan untuk menjalankan kueri berkelanjutan di seluruh project, Anda harus memperbarui kebijakan organisasi untuk mengizinkan pemasangan akun layanan dalam satu project ke resource di project lain. Untuk mengetahui informasi selengkapnya, lihat Mengaktifkan akun layanan untuk dilampirkan di seluruh project.

Ikuti langkah-langkah berikut untuk menggunakan akun layanan guna menjalankan kueri berkelanjutan:

Konsol

  1. Membuat akun layanan:
  2. Berikan izin yang diperlukan ke akun layanan.

  3. Di konsol Google Cloud , buka halaman BigQuery.

    Buka BigQuery

  4. Di editor kueri, klik Edit > Query mode > Kueri berkelanjutan. Untuk mengonfirmasi pilihan ini, klik Konfirmasi.

  5. Di editor kueri, klik Edit > Query settings.

  6. Di bagian Continuous query, gunakan kotak Service account untuk memilih akun layanan yang Anda buat.

  7. Opsional: Untuk mengontrol durasi kueri berjalan, tetapkan Waktu tunggu tugas dalam milidetik.

  8. Klik Simpan.

  9. Di editor kueri, ketik pernyataan SQL untuk kueri berkelanjutan. Pernyataan SQL hanya boleh berisi operasi yang didukung.

  10. Klik Run.

bq

  1. Membuat akun layanan:
  2. Berikan izin yang diperlukan ke akun layanan.

  3. Di konsol Google Cloud , aktifkan Cloud Shell.

    Aktifkan Cloud Shell

    Di bagian bawah konsol Google Cloud , sesi Cloud Shell akan dimulai dan menampilkan prompt command line. Cloud Shell adalah lingkungan shell dengan Google Cloud CLI yang sudah terinstal, dan dengan nilai yang sudah ditetapkan untuk project Anda saat ini. Diperlukan waktu beberapa detik untuk melakukan inisialisasi pada sesi.

  4. Pada command line, jalankan kueri berkelanjutan menggunakan perintah bq query dengan flag berikut:

    • Tetapkan flag --continuous ke true untuk membuat kueri berkelanjutan.
    • Gunakan flag --connection_property untuk menentukan akun layanan yang akan digunakan.
    • Opsional: Tetapkan flag --job_timeout_ms untuk membatasi runtime kueri.
    bq query --project_id=PROJECT_ID --use_legacy_sql=false \
--continuous=true --connection_property=service_account=SERVICE_ACCOUNT_EMAIL \
'QUERY'

    Ganti kode berikut:

    • PROJECT_ID: project ID Anda.
    • SERVICE_ACCOUNT_EMAIL: email akun layanan. Anda bisa mendapatkan email akun layanan dari halaman Service accounts di konsol Google Cloud .
    • QUERY: pernyataan SQL untuk kueri berkelanjutan. Pernyataan SQL hanya boleh berisi operasi yang didukung.

API

  1. Membuat akun layanan:
  2. Berikan izin yang diperlukan ke akun layanan.

  3. Jalankan kueri berkelanjutan dengan memanggil metode jobs.insert. Tetapkan kolom berikut di resource JobConfigurationQuery dari resource Job yang Anda teruskan:

    • Tetapkan kolom continuous ke true untuk membuat kueri berkelanjutan.
    • Gunakan kolom connectionProperties untuk menentukan akun layanan yang akan digunakan.

    Anda dapat mengontrol durasi kueri berjalan dengan menetapkan kolom jobTimeoutMs di resource JobConfiguration.

    curl --request POST \
  "https://bigquery.googleapis.com/bigquery/v2/projects/PROJECT_ID/jobs" \
  --header "Authorization: Bearer $(gcloud auth print-access-token)" \
  --header "Content-Type: application/json; charset=utf-8" \
  --data '{"configuration":{"query":{"query":"QUERY","useLegacySql":false,"continuous":true,"connectionProperties":[{"key":"service_account","value":"SERVICE_ACCOUNT_EMAIL"}]}}}' \
  --compressed

    Ganti kode berikut:

    • PROJECT_ID: project ID Anda.
    • QUERY: pernyataan SQL untuk kueri berkelanjutan. Pernyataan SQL hanya boleh berisi operasi yang didukung.
    • SERVICE_ACCOUNT_EMAIL: email akun layanan. Anda bisa mendapatkan email akun layanan di halaman Service accounts di konsol Google Cloud .

Membuat ID pekerjaan kustom

Setiap tugas kueri diberi ID pekerjaan yang dapat Anda gunakan untuk menelusuri dan mengelola tugas. Secara default, ID tugas dibuat secara acak. Agar lebih mudah menelusuri ID pekerjaan kueri berkelanjutan menggunakan histori tugas atau penjelajah tugas, Anda dapat menetapkan awalan ID pekerjaan kustom:

  1. Di konsol Google Cloud , buka halaman BigQuery.

    Buka BigQuery

  2. Di editor kueri, klik Edit > Mode kueri > Kueri berkelanjutan.

  3. Klik Konfirmasi.

  4. Di editor kueri, klik Edit > Query settings.

  5. Di bagian Custom job ID prefix, masukkan awalan ID pekerjaan kustom.

  6. Klik Simpan.

Pemrosesan stateful dengan JOIN dan agregasi windowing

Operasi stateful memungkinkan kueri berkelanjutan melakukan analisis kompleks dengan mempertahankan informasi di beberapa baris atau interval waktu. Operasi ini mencakup JOIN dan agregasi windowing.

Untuk mengetahui informasi mendetail tentang cara menggunakan operasi stateful ini, lihat topik berikut:

  • JOIN kueri berkelanjutan melakukan korelasi real-time antara beberapa aliran data berorientasi waktu.
  • Agregasi windowing mengelompokkan data streaming ke dalam interval waktu yang konsisten untuk analisis menggunakan fungsi agregasi.

Contoh

Contoh SQL berikut menunjukkan kasus penggunaan umum untuk kueri berkelanjutan.

Mengekspor data ke topik Pub/Sub

Contoh berikut menunjukkan kueri berkelanjutan yang memfilter data dari tabel BigQuery yang menerima informasi perjalanan taksi streaming, dan memublikasikan data untuk perjalanan yang dibatalkan ke topik Pub/Sub secara real time dengan atribut pesan:

EXPORT DATA
  OPTIONS (
    format = 'CLOUD_PUBSUB',
    uri = 'https://pubsub.googleapis.com/projects/myproject/topics/taxi-real-time-rides')
AS (
  SELECT
    TO_JSON_STRING(
      STRUCT(
        ride_id,
        timestamp,
        latitude,
        longitude)) AS message,
    TO_JSON(
      STRUCT(
        CAST(passenger_comment AS STRING) AS passenger_comment))
  FROM
    CHANGES(TABLE `myproject.real_time_taxi_streaming.taxi_rides`,
      -- Configure the CHANGES TVF start_timestamp to specify when you want to
      -- start processing data using your continuous query.
      -- This example starts processing at 10 minutes before the current time.
      CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE _CHANGE_TYPE = 'DELETE'
);

Mengekspor data ke tabel Bigtable

Contoh berikut menunjukkan kueri berkelanjutan yang memfilter data dari tabel BigQuery yang menerima informasi perjalanan taksi streaming, dan mengekspor data ke tabel Bigtable secara real time:

EXPORT DATA
  OPTIONS (
    format = 'CLOUD_BIGTABLE',
    truncate = TRUE,
    overwrite = TRUE,
    uri = 'https://bigtable.googleapis.com/projects/myproject/instances/mybigtableinstance/tables/taxi-real-time-rides')
AS (
  SELECT
    CAST(CONCAT(ride_id, timestamp, latitude, longitude) AS STRING) AS rowkey,
    STRUCT(
      timestamp,
      latitude,
      longitude,
      meter_reading,
      ride_status,
      passenger_count) AS features
  FROM
    APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
      -- Configure the APPENDS TVF start_timestamp to specify when you want to
      -- start processing data using your continuous query.
      -- This example starts processing at 10 minutes before the current time.
      CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE ride_status = 'enroute'
);

Mengekspor data ke tabel Spanner

Contoh berikut menunjukkan kueri berkelanjutan yang memfilter data dari tabel BigQuery yang menerima informasi perjalanan taksi streaming, lalu mengekspor data ke tabel Spanner secara real time:

EXPORT DATA
 OPTIONS (
   format = 'CLOUD_SPANNER',
   uri = 'https://spanner.googleapis.com/projects/myproject/instances/myspannerinstance/databases/taxi-real-time-rides',
   spanner_options ="""{
      "table": "rides",
      -- To ensure data is written to Spanner in the correct sequence
      -- during a continuous export, use the change_timestamp_column
      -- option. This should be mapped to a timestamp column from your
      -- BigQuery data. If your source data lacks a timestamp, the
      -- _CHANGE_TIMESTAMP pseudocolumn provided by the APPENDS function
      -- will be automatically mapped to the "change_timestamp" column.
      "change_timestamp_column": "change_timestamp"
   }"""
  )
  AS (
  SELECT
    ride_id,
    latitude,
    longitude,
    meter_reading,
    ride_status,
    passenger_count,
    _CHANGE_TIMESTAMP as change_timestamp
  FROM APPENDS(
        TABLE `myproject.real_time_taxi_streaming.taxirides`,
        -- Configure the APPENDS TVF start_timestamp to specify when you want to
        -- start processing data using your continuous query.
        -- This example starts processing at 10 minutes before the current time.
        CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
  WHERE ride_status = 'enroute'
  );

Menulis data ke tabel BigQuery atau tabel terkelola Apache Iceberg

Contoh berikut menunjukkan kueri berkelanjutan yang memfilter dan mengubah data dari tabel BigQuery yang menerima informasi perjalanan taksi streaming, lalu menulis data ke tabel BigQuery lain atau tabel terkelola Iceberg secara real time. Hal ini membuat data tersedia untuk analisis downstream lebih lanjut di BigQuery atau lakehouse format terbuka Anda.

INSERT INTO `myproject.real_time_taxi_streaming.transformed_taxirides`
SELECT
  timestamp,
  meter_reading,
  ride_status,
  passenger_count,
  ST_Distance(
    ST_GeogPoint(pickup_longitude, pickup_latitude),
    ST_GeogPoint(dropoff_longitude, dropoff_latitude)) AS euclidean_trip_distance,
    SAFE_DIVIDE(meter_reading, passenger_count) AS cost_per_passenger
FROM
  APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
    -- Configure the APPENDS TVF start_timestamp to specify when you want to
    -- start processing data using your continuous query.
    -- This example starts processing at 10 minutes before the current time.
    CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
WHERE
  ride_status = 'dropoff';

Untuk menulis output kueri berkelanjutan ke tabel terkelola Iceberg, ganti myproject.real_time_taxi_streaming.transformed_taxirides dengan nama tabel terkelola Iceberg target Anda. Sintaksisnya identik karena detail koneksi dan setelan penyimpanan lakehouse ditentukan di tabel tujuan itu sendiri.

Memproses data menggunakan model Gemini Enterprise Agent Platform

Contoh berikut menunjukkan kueri berkelanjutan yang menggunakan model Agent Platform untuk membuat iklan bagi penumpang taksi berdasarkan lintang dan bujur mereka saat ini, lalu mengekspor hasilnya ke topik Pub/Sub secara real time:

EXPORT DATA
  OPTIONS (
    format = 'CLOUD_PUBSUB',
    uri = 'https://pubsub.googleapis.com/projects/myproject/topics/taxi-real-time-rides')
AS (
  SELECT
    TO_JSON_STRING(
      STRUCT(
        ride_id,
        timestamp,
        latitude,
        longitude,
        prompt,
        result)) AS message
  FROM
    AI.GENERATE_TEXT(
      MODEL `myproject.real_time_taxi_streaming.taxi_ml_generate_model`,
      (
        SELECT
          timestamp,
          ride_id,
          latitude,
          longitude,
          CONCAT(
            'Generate an ad based on the current latitude of ',
            latitude,
            ' and longitude of ',
            longitude) AS prompt
        FROM
          APPENDS(TABLE `myproject.real_time_taxi_streaming.taxirides`,
            -- Configure the APPENDS TVF start_timestamp to specify when you
            -- want to start processing data using your continuous query.
            -- This example starts processing at 10 minutes before the current time.
            CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE)
        WHERE ride_status = 'enroute'
      ),
      STRUCT(
        50 AS max_output_tokens,
        1.0 AS temperature,
        40 AS top_k,
        1.0 AS top_p))
      AS ml_output
);

Melakukan agregasi JOIN dan windowing

Contoh berikut menunjukkan kueri berkelanjutan yang melakukan JOIN dan agregasi windowing.

Misalkan Anda ingin menggabungkan tabel perjalanan taksi ke tabel permintaan taksi untuk memahami kesehatan taksi di setiap lingkungan setiap lima menit. Dengan menggunakan fungsi agregat, Anda dapat menangkap volume permintaan taksi per lingkungan dan jarak minimum, maksimum, rata-rata, dan simpangan baku antara penumpang dan taksi saat mereka meminta tumpangan.

INSERT INTO
 `real_time_taxi_streaming.neighborhood_taxi_health`
WITH potential_matches AS (
 SELECT
   requests._CHANGE_TIMESTAMP AS bq_changed_ts,
   requests.geohash,
   requests.latitude,
   requests.longitude,
   ST_DISTANCE(
     ST_GEOGPOINT(requests.longitude, requests.latitude),
     ST_GEOGPOINT(taxis.longitude, taxis.latitude)
   ) AS distance_in_meters
 FROM
   APPENDS(TABLE `real_time_taxi_streaming.ride_requests`,
     CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) AS requests
 INNER JOIN
   APPENDS(TABLE `real_time_taxi_streaming.taxirides`,
     CURRENT_TIMESTAMP() - INTERVAL 10 MINUTE) AS taxis
 ON requests.geohash = taxis.geohash
 WHERE
   taxis.ride_status = 'available'
   AND taxis._CHANGE_TIMESTAMP BETWEEN (requests._CHANGE_TIMESTAMP - INTERVAL 5 MINUTE) AND requests._CHANGE_TIMESTAMP
   AND ST_Dwithin(
     ST_GEOGPOINT(requests.longitude, requests.latitude),
     ST_GEOGPOINT(taxis.longitude, taxis.latitude),
     2000 -- Distance in meters
   )
)
SELECT
 window_end,
 geohash,
 ROUND(AVG(latitude), 6) AS avg_latitude,
 ROUND(AVG(longitude), 6) AS avg_longitude,
 COUNT(*) AS taxi_demand_volume,
 ROUND(AVG(distance_in_meters), 2) AS avg_proximity_meters,
 ROUND(MIN(distance_in_meters), 2) AS min_proximity_meters,
 ROUND(MAX(distance_in_meters), 2) AS max_proximity_meters,
 ROUND(STDDEV(distance_in_meters), 2) AS proximity_stddev
FROM
 TUMBLE(TABLE potential_matches, "bq_changed_ts", INTERVAL 5 MINUTE)
GROUP BY
 window_end,
 geohash;

Mengubah SQL kueri berkelanjutan

Anda tidak dapat memperbarui SQL yang digunakan dalam kueri berkelanjutan saat tugas kueri berkelanjutan sedang berjalan. Anda harus membatalkan tugas kueri berkelanjutan, mengubah SQL, lalu memulai tugas kueri berkelanjutan baru dari titik tempat Anda menghentikan tugas kueri berkelanjutan asli.

Ikuti langkah-langkah berikut untuk mengubah SQL yang digunakan dalam kueri berkelanjutan:

  1. Lihat detail tugas untuk tugas kueri berkelanjutan yang ingin Anda perbarui, dan catat ID pekerjaan.
  2. Jika memungkinkan, jeda pengumpulan data upstream. Jika tidak dapat melakukannya, Anda mungkin mendapatkan beberapa duplikasi data saat kueri berkelanjutan dimulai ulang.
  3. Batalkan kueri berkelanjutan yang ingin Anda ubah.

  4. Dapatkan nilai end_time untuk tugas kueri berkelanjutan asli menggunakan tampilan JOBS INFORMATION_SCHEMA:

    SELECT end_time
FROM `PROJECT_ID.region-REGION`.INFORMATION_SCHEMA.JOBS_BY_PROJECT
WHERE
  EXTRACT(DATE FROM creation_time) = current_date()
AND error_result.reason = 'stopped'
AND job_id = 'JOB_ID';

    Ganti kode berikut:

    • PROJECT_ID: project ID Anda.
    • REGION: region yang digunakan oleh project Anda.
    • JOB_ID: ID pekerjaan kueri berkelanjutan yang Anda identifikasi di Langkah 1.

  5. Ubah pernyataan SQL kueri berkelanjutan untuk memulai kueri berkelanjutan dari titik waktu tertentu, menggunakan nilai end_time yang Anda ambil di Langkah 5 sebagai titik awal.

  6. Ubah pernyataan SQL kueri berkelanjutan untuk mencerminkan perubahan yang Anda perlukan.

  7. Jalankan kueri berkelanjutan yang telah diubah.

Membatalkan kueri berkelanjutan

Anda dapat membatalkan tugas kueri berkelanjutan seperti tugas lainnya. Mungkin perlu waktu hingga satu menit agar kueri berhenti berjalan setelah tugas dibatalkan.

Jika Anda membatalkan, lalu memulai ulang kueri, kueri yang dimulai ulang akan berperilaku seperti kueri baru yang independen. Kueri yang dimulai ulang tidak mulai memproses data di tempat tugas sebelumnya berhenti dan tidak dapat mereferensikan hasil kueri sebelumnya. Lihat Memulai kueri berkelanjutan dari titik waktu tertentu.

Memantau kueri dan menangani error

Kueri berkelanjutan mungkin terganggu karena faktor-faktor seperti inkonsistensi data, perubahan skema, gangguan layanan sementara, atau pemeliharaan. Meskipun BigQuery menangani beberapa error sementara, praktik terbaik untuk meningkatkan ketahanan tugas mencakup hal berikut:

Langkah berikutnya