Menggunakan konektor BigQuery dengan Dataproc Serverless untuk Spark

Gunakan spark-bigquery-connector dengan Apache Spark untuk membaca dan menulis data dari dan ke BigQuery. Tutorial ini menunjukkan aplikasi PySpark yang menggunakan spark-bigquery-connector.

Menggunakan konektor BigQuery dengan beban kerja Anda

Lihat Dataproc Serverless untuk rilis runtime Spark untuk menentukan versi konektor BigQuery yang terinstal di versi runtime beban kerja batch Anda. Jika konektor tidak tercantum, lihat bagian berikutnya untuk mengetahui petunjuk cara menyediakan konektor untuk aplikasi.

Cara menggunakan konektor dengan runtime Spark versi 2.0

Konektor BigQuery tidak diinstal di runtime Spark versi 2.0. Saat menggunakan Spark runtime versi 2.0, Anda dapat menyediakan konektor untuk aplikasi Anda dengan salah satu cara berikut:

  • Gunakan parameter jars untuk mengarah ke file jar konektor saat Anda mengirimkan Dataproc Serverless untuk workload batch Spark. Contoh berikut menentukan file jar konektor (lihat repositori GoogleCloudDataproc/spark-bigquery-connector di GitHub untuk mengetahui daftar file jar konektor yang tersedia).
    • Contoh Google Cloud CLI: 
      gcloud dataproc batches submit pyspark \
    --region=region \
    --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.13-version.jar \
    ... other args
  • Sertakan file jar konektor dalam aplikasi Spark sebagai dependensi (lihat Melakukan kompilasi terhadap konektor)

Menghitung biaya

Tutorial ini menggunakan komponen Google Cloud yang dapat ditagih, termasuk:

  • Serverless Dataproc
  • BigQuery
  • Cloud Storage

Gunakan Kalkulator Harga untuk membuat perkiraan biaya berdasarkan penggunaan yang Anda proyeksikan. Pengguna Cloud Platform baru mungkin memenuhi syarat untuk mendapatkan uji coba gratis.

BigQuery I/O

Contoh ini membaca data dari BigQuery ke Spark DataFrame untuk melakukan jumlah kata menggunakan API sumber data standar.

Konektor menulis output jumlah kata ke BigQuery dengan:

  1. Membuat buffer data menjadi file sementara di bucket Cloud Storage

  2. Menyalin data dalam satu operasi dari bucket Cloud Storage Anda ke BigQuery

  3. Menghapus file sementara di Cloud Storage setelah operasi pemuatan BigQuery selesai (file sementara juga akan dihapus setelah aplikasi Spark dihentikan). Jika penghapusan gagal, Anda harus menghapus file Cloud Storage sementara yang tidak diinginkan, yang biasanya ditempatkan di gs://your-bucket/.spark-bigquery-jobid-UUID.

Konfigurasi penagihan

Secara default, project yang terkait dengan kredensial atau akun layanan ditagih untuk penggunaan API. Untuk menagih project yang berbeda, tetapkan konfigurasi berikut: spark.conf.set("parentProject", "<BILLED-GCP-PROJECT>").

Ini juga dapat ditambahkan ke operasi baca/tulis, seperti berikut: .option("parentProject", "<BILLED-GCP-PROJECT>").

Mengirimkan workload batch jumlah kata PySpark

  1. Buat wordcount_dataset dengan alat command line bq di terminal lokal atau di Cloud Shell. 
    bq mk wordcount_dataset
  2. Buat bucket Cloud Storage dengan alat command line gsutil" di terminal lokal atau di Cloud Shell. 
    gsutil mb gs://your-bucket
  3. Periksa kodenya. 
    #!/usr/bin/python
"""BigQuery I/O PySpark example."""
from pyspark.sql import SparkSession

spark = SparkSession \
  .builder \
  .appName('spark-bigquery-demo') \
  .getOrCreate()

# Use the Cloud Storage bucket for temporary BigQuery export data used
# by the connector.
bucket = "[your-bucket-name]"
spark.conf.set('temporaryGcsBucket', bucket)

# Load data from BigQuery.
words = spark.read.format('bigquery') \
  .option('table', 'bigquery-public-data:samples.shakespeare') \
  .load()
words.createOrReplaceTempView('words')

# Perform word count.
word_count = spark.sql(
    'SELECT word, SUM(word_count) AS word_count FROM words GROUP BY word')
word_count.show()
word_count.printSchema()

# Saving the data to BigQuery
word_count.write.format('bigquery') \
  .option('table', 'wordcount_dataset.wordcount_output') \
  .save()
  4. Buat wordcount.py secara lokal di editor teks dengan menyalin kode PySpark dari listingan kode PySpark. Ganti placeholder [your-bucket] dengan nama bucket Cloud Storage yang Anda buat.
  5. Kirim workload batch PySpark: 
    gcloud dataproc batches submit pyspark wordcount.py \
    --region=region \
    --deps-bucket=your-bucket
    Contoh output terminal: 
    ...
+---------+----------+
|     word|word_count|
+---------+----------+
|     XVII|         2|
|    spoil|        28|
|    Drink|         7|
|forgetful|         5|
|   Cannot|        46|
|    cures|        10|
|   harder|        13|
|  tresses|         3|
|      few|        62|
|  steel'd|         5|
| tripping|         7|
|   travel|        35|
|   ransom|        55|
|     hope|       366|
|       By|       816|
|     some|      1169|
|    those|       508|
|    still|       567|
|      art|       893|
|    feign|        10|
+---------+----------+
only showing top 20 rows

root
 |-- word: string (nullable = false)
 |-- word_count: long (nullable = true)

    Untuk melihat pratinjau tabel output di Google Cloud Console, buka halaman BigQuery project Anda, pilih tabel wordcount_output, lalu klik Pratinjau.

Untuk informasi selengkapnya