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
- Contoh Google Cloud CLI:
- 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:
Membuat buffer data menjadi file sementara di bucket Cloud Storage
Menyalin data dalam satu operasi dari bucket Cloud Storage Anda ke BigQuery
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
- Buat
wordcount_dataset
dengan alat command line bq di terminal lokal atau di Cloud Shell.bq mk wordcount_dataset
- Buat bucket Cloud Storage dengan
alat command line gsutil" di
terminal lokal atau di
Cloud Shell.
gsutil mb gs://your-bucket
- 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()
- 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. - 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 tabelwordcount_output
, lalu klik Pratinjau.
Untuk informasi selengkapnya
- BigQuery Storage & Spark SQL - Python
- Membuat file definisi tabel untuk sumber data eksternal
- Membuat kueri data yang dipartisi secara eksternal