Halaman ini diterjemahkan oleh Cloud Translation API.

Menggunakan Dataproc, BigQuery, dan Apache Spark ML untuk Machine Learning

BigQuery Connector untuk Apache Spark memungkinkan Data Scientist memadukan kecanggihan teknologi BigQuery SQL engine yang skalabel tanpa hambatan dengan Machine Learning Apache Spark kemampuan IT. Dalam tutorial ini, kami menunjukkan cara menggunakan Dataproc, BigQuery dan Apache Spark ML untuk menjalankan machine learning pada set data.

Tujuan

Gunakan regresi linier untuk membangun model berat lahir sebagai fungsi dari lima faktor:

minggu kehamilan
usia ibu
usia ayah
pertambahan berat ibu selama kehamilan
Skor Apgar

Gunakan alat berikut:

BigQuery, untuk menyiapkan tabel input regresi linear, yaitu yang ditulis ke project Google Cloud Anda
Python, untuk membuat kueri dan mengelola data di BigQuery
Apache Spark, untuk mengakses tabel regresi linear yang dihasilkan
Spark ML, untuk membangun dan mengevaluasi model
Tugas Dataproc PySpark, untuk memanggil fungsi Spark ML

Biaya

Dalam dokumen ini, Anda menggunakan komponen Google Cloud yang dapat ditagih berikut:

Compute Engine
Dataproc
BigQuery

Untuk membuat perkiraan biaya berdasarkan proyeksi penggunaan Anda, gunakan kalkulator harga. Pengguna baru Google Cloud mungkin memenuhi syarat untuk mendapatkan uji coba gratis.

Sebelum memulai

Cluster Dataproc memiliki komponen Spark, termasuk Spark ML, yang terinstal. Untuk menyiapkan cluster Dataproc dan menjalankan kode dalam contoh ini, Anda akan perlu melakukan (atau telah melakukan) hal berikut:

Login ke akun Google Cloud Anda. Jika Anda baru menggunakan Google Cloud, buat akun untuk mengevaluasi performa produk kami dalam skenario dunia nyata. Pelanggan baru juga mendapatkan kredit gratis senilai $300 untuk menjalankan, menguji, dan men-deploy workload.

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

Buka pemilih project

Aktifkan API Dataproc, BigQuery, Compute Engine.

Mengaktifkan API

Menginstal Google Cloud CLI.

Untuk initialize gcloud CLI, jalankan perintah berikut:

gcloud init

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

Buka pemilih project

Aktifkan API Dataproc, BigQuery, Compute Engine.

Mengaktifkan API

Menginstal Google Cloud CLI.

Untuk initialize gcloud CLI, jalankan perintah berikut:

gcloud init

Buat cluster Dataproc di proyek. Cluster Anda harus menjalankan Versi Dataproc dengan Spark 2.0 atau yang lebih tinggi, (termasuk library machine learning).

Membuat Subkumpulan data `natality` BigQuery

Di bagian ini, Anda akan membuat set data di project Anda, lalu membuat tabel tempat Anda menyalin {i>subset<i} data tingkat kelahiran dari kelahiran yang tersedia untuk umum set data BigQuery. Nanti dalam tutorial ini, Anda akan menggunakan {i>subset data<i} dalam tabel ini untuk memprediksi berat lahir sebagai fungsi usia ibu, usia ayah, dan minggu kehamilan.

Anda dapat membuat {i>subset <i}data menggunakan Konsol Google Cloud atau menjalankan skrip Python di komputer lokal Anda.

Konsol

Buat set data di project Anda.
1. Buka UI Web BigQuery.
2. Di panel navigasi kiri, klik nama proyek Anda, lalu klik BUAT SET DATA.
3. Dalam dialog Create dataset:
  1. Untuk Dataset ID, masukkan "natality_regression".
  2. Untuk Lokasi data, Anda dapat memilih lokasi untuk set data. Lokasi nilai default adalah US multi-region. Setelah set data dibuat, lokasi tidak dapat diubah.
  3. Untuk Masa berlaku tabel default, pilih salah satu opsi berikut:
    - Never (default): Anda harus menghapus tabel secara manual.
    - Jumlah hari: Tabel akan dihapus setelah jumlah hari yang ditentukan sejak waktu pembuatannya.
  4. Untuk Enkripsi, pilih salah satu opsi berikut:
    - Kunci yang dimiliki Google dan dikelola Google (default).
    - Kunci yang dikelola pelanggan: Lihat daftar Melindungi data dengan kunci Cloud KMS.
  5. Klik Create dataset.
    Anda tidak dapat menambahkan deskripsi atau label saat membuat set data menggunakan UI Web. Setelah set data dibuat, Anda bisa menambahkan deskripsi, dan tambahkan label.

Jalankan kueri terhadap {i>dataset<i} kelahiran publik, lalu simpan kueri tersebut menghasilkan tabel baru dalam {i>dataset<i} Anda.

Salin dan tempel kueri berikut ke dalam Editor Kueri, lalu klik Run.

CREATE OR REPLACE TABLE natality_regression.regression_input as
SELECT
weight_pounds,
mother_age,
father_age,
gestation_weeks,
weight_gain_pounds,
apgar_5min
FROM
`bigquery-public-data.samples.natality`
WHERE
weight_pounds IS NOT NULL
AND mother_age IS NOT NULL
AND father_age IS NOT NULL
AND gestation_weeks IS NOT NULL
AND weight_gain_pounds IS NOT NULL
AND apgar_5min IS NOT NULL

Setelah kueri selesai (dalam waktu sekitar satu menit), hasilnya disimpan sebagai "regression_input" Tabel BigQuery dalam set data natality_regression di project Anda.

Python

Sebelum mencoba contoh ini, ikuti petunjuk penyiapan Python di Panduan memulai Dataproc menggunakan library klien. Untuk informasi selengkapnya, lihat API Python Dataproc dokumentasi referensi.

Untuk mengautentikasi ke Dataproc, siapkan Kredensial Default Aplikasi. Untuk informasi selengkapnya, lihat Menyiapkan autentikasi untuk lingkungan pengembangan lokal.

Lihat Menyiapkan Lingkungan Pengembangan Python untuk mendapatkan petunjuk tentang cara menginstal Python dan Library Klien Google Cloud untuk Python (diperlukan untuk menjalankan kode). Menginstal dan menggunakan Python virtualenv direkomendasikan.

Salin dan tempel kode natality_tutorial.py, di bawah, ke dalam shell python di komputer lokal Anda. Tekan tombol <return> di shell untuk menjalankan kode guna membuat "natality_regression" set data BigQuery di setelan default project Google Cloud dengan "regression_input" tabel yang diisi dengan subset publik Data natality.

"""Create a Google BigQuery linear regression input table.

In the code below, the following actions are taken:
* A new dataset is created "natality_regression."
* A query is run against the public dataset,
    bigquery-public-data.samples.natality, selecting only the data of
    interest to the regression, the output of which is stored in a new
    "regression_input" table.
* The output table is moved over the wire to the user's default project via
    the built-in BigQuery Connector for Spark that bridges BigQuery and
    Cloud Dataproc.
"""

from google.cloud import bigquery

# Create a new Google BigQuery client using Google Cloud Platform project
# defaults.
client = bigquery.Client()

# Prepare a reference to a new dataset for storing the query results.
dataset_id = "natality_regression"
dataset_id_full = f"{client.project}.{dataset_id}"

dataset = bigquery.Dataset(dataset_id_full)

# Create the new BigQuery dataset.
dataset = client.create_dataset(dataset)

# Configure the query job.
job_config = bigquery.QueryJobConfig()

# Set the destination table to where you want to store query results.
# As of google-cloud-bigquery 1.11.0, a fully qualified table ID can be
# used in place of a TableReference.
job_config.destination = f"{dataset_id_full}.regression_input"

# Set up a query in Standard SQL, which is the default for the BigQuery
# Python client library.
# The query selects the fields of interest.
query = """
    SELECT
        weight_pounds, mother_age, father_age, gestation_weeks,
        weight_gain_pounds, apgar_5min
    FROM
        `bigquery-public-data.samples.natality`
    WHERE
        weight_pounds IS NOT NULL
        AND mother_age IS NOT NULL
        AND father_age IS NOT NULL
        AND gestation_weeks IS NOT NULL
        AND weight_gain_pounds IS NOT NULL
        AND apgar_5min IS NOT NULL
"""

# Run the query.
client.query_and_wait(query, job_config=job_config)  # Waits for the query to finish

Konfirmasi pembuatan set data natality_regression dan tabel regression_input.

Menjalankan regresi linear

Di bagian ini, Anda akan menjalankan regresi linear PySpark dengan mengirimkan tugas ke layanan Dataproc menggunakan konsol Google Cloud atau dengan menjalankan perintah gcloud dari terminal lokal.

Konsol

Salin dan tempel kode berikut ke natality_sparkml.py baru file di komputer lokal Anda.

"""Run a linear regression using Apache Spark ML.

In the following PySpark (Spark Python API) code, we take the following actions:

  * Load a previously created linear regression (BigQuery) input table
    into our Cloud Dataproc Spark cluster as an RDD (Resilient
    Distributed Dataset)
  * Transform the RDD into a Spark Dataframe
  * Vectorize the features on which the model will be trained
  * Compute a linear regression using Spark ML

"""
from pyspark.context import SparkContext
from pyspark.ml.linalg import Vectors
from pyspark.ml.regression import LinearRegression
from pyspark.sql.session import SparkSession
# The imports, above, allow us to access SparkML features specific to linear
# regression as well as the Vectors types.


# Define a function that collects the features of interest
# (mother_age, father_age, and gestation_weeks) into a vector.
# Package the vector in a tuple containing the label (`weight_pounds`) for that
# row.
def vector_from_inputs(r):
  return (r["weight_pounds"], Vectors.dense(float(r["mother_age"]),
                                            float(r["father_age"]),
                                            float(r["gestation_weeks"]),
                                            float(r["weight_gain_pounds"]),
                                            float(r["apgar_5min"])))

sc = SparkContext()
spark = SparkSession(sc)

# Read the data from BigQuery as a Spark Dataframe.
natality_data = spark.read.format("bigquery").option(
    "table", "natality_regression.regression_input").load()
# Create a view so that Spark SQL queries can be run against the data.
natality_data.createOrReplaceTempView("natality")

# As a precaution, run a query in Spark SQL to ensure no NULL values exist.
sql_query = """
SELECT *
from natality
where weight_pounds is not null
and mother_age is not null
and father_age is not null
and gestation_weeks is not null
"""
clean_data = spark.sql(sql_query)

# Create an input DataFrame for Spark ML using the above function.
training_data = clean_data.rdd.map(vector_from_inputs).toDF(["label",
                                                             "features"])
training_data.cache()

# Construct a new LinearRegression object and fit the training data.
lr = LinearRegression(maxIter=5, regParam=0.2, solver="normal")
model = lr.fit(training_data)
# Print the model summary.
print("Coefficients:" + str(model.coefficients))
print("Intercept:" + str(model.intercept))
print("R^2:" + str(model.summary.r2))
model.summary.residuals.show()

Menyalin file natality_sparkml.py lokal ke Cloud Storage bucket tertentu di project Anda.
```
gcloud storage cp natality_sparkml.py gs://bucket-name
```
Alih-alih menyalin file ke bucket pengguna dalam project, Anda dapat salin ke bucket staging yang dibuat Dataproc saat Anda membuat .
Menjalankan regresi dari Dataproc Halaman Mengirim tugas.
1. Di kolom Main python file, masukkan URI gs:// dari bucket Cloud Storage tempat salinan natality_sparkml.py Anda file tersebut ditemukan.
2. Pilih PySpark sebagai Jenis pekerjaan.
3. Sisipkan gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar di kolom Jar files. Hal ini membuat konektor spark-bigquery-tersedia ke aplikasi PySpark saat runtime agar aplikasi PySpark dapat membaca BigQuery data ke dalam Spark DataFrame.
  Jar 2.12 kompatibel dengan cluster Dataproc yang dibuat dengan gambar versi 1.5 atau yang lebih baru. Jika cluster Dataproc Anda dibuat dengan Gambar 1.3 atau 1.4, tentukan jar 2.11 sebagai gantinya (gs://spark-lib/bigquery/spark-bigquery-latest_2.11.jar).
4. Isi kolom Job ID, Region, dan Cluster.
5. Klik Submit untuk menjalankan tugas di cluster Anda.

Setelah tugas selesai, ringkasan model output regresi linear muncul di jendela detail Tugas Dataproc.

gcloud

Salin dan tempel kode berikut ke natality_sparkml.py baru file di komputer lokal Anda.

"""Run a linear regression using Apache Spark ML.

In the following PySpark (Spark Python API) code, we take the following actions:

  * Load a previously created linear regression (BigQuery) input table
    into our Cloud Dataproc Spark cluster as an RDD (Resilient
    Distributed Dataset)
  * Transform the RDD into a Spark Dataframe
  * Vectorize the features on which the model will be trained
  * Compute a linear regression using Spark ML

"""
from pyspark.context import SparkContext
from pyspark.ml.linalg import Vectors
from pyspark.ml.regression import LinearRegression
from pyspark.sql.session import SparkSession
# The imports, above, allow us to access SparkML features specific to linear
# regression as well as the Vectors types.


# Define a function that collects the features of interest
# (mother_age, father_age, and gestation_weeks) into a vector.
# Package the vector in a tuple containing the label (`weight_pounds`) for that
# row.
def vector_from_inputs(r):
  return (r["weight_pounds"], Vectors.dense(float(r["mother_age"]),
                                            float(r["father_age"]),
                                            float(r["gestation_weeks"]),
                                            float(r["weight_gain_pounds"]),
                                            float(r["apgar_5min"])))

sc = SparkContext()
spark = SparkSession(sc)

# Read the data from BigQuery as a Spark Dataframe.
natality_data = spark.read.format("bigquery").option(
    "table", "natality_regression.regression_input").load()
# Create a view so that Spark SQL queries can be run against the data.
natality_data.createOrReplaceTempView("natality")

# As a precaution, run a query in Spark SQL to ensure no NULL values exist.
sql_query = """
SELECT *
from natality
where weight_pounds is not null
and mother_age is not null
and father_age is not null
and gestation_weeks is not null
"""
clean_data = spark.sql(sql_query)

# Create an input DataFrame for Spark ML using the above function.
training_data = clean_data.rdd.map(vector_from_inputs).toDF(["label",
                                                             "features"])
training_data.cache()

# Construct a new LinearRegression object and fit the training data.
lr = LinearRegression(maxIter=5, regParam=0.2, solver="normal")
model = lr.fit(training_data)
# Print the model summary.
print("Coefficients:" + str(model.coefficients))
print("Intercept:" + str(model.intercept))
print("R^2:" + str(model.summary.r2))
model.summary.residuals.show()

Menyalin file natality_sparkml.py lokal ke Cloud Storage bucket tertentu di project Anda.
```
gcloud storage cp natality_sparkml.py gs://bucket-name
```
Alih-alih menyalin file ke bucket pengguna dalam project, Anda dapat salin ke bucket staging yang dibuat Dataproc saat Anda membuat .
Kirim tugas Pyspark ke layanan Dataproc dengan menjalankan Perintah gcloud, yang ditampilkan di bawah, dari jendela terminal di komputer lokal Anda.
1. Nilai flag --jars akan menyediakan spark-bigquery-connector ke tugas PySpark saat runtime agar tugas dapat dibaca Data BigQuery ke dalam Spark DataFrame.
```
gcloud dataproc jobs submit pyspark \
    gs://your-bucket/natality_sparkml.py \
    --cluster=cluster-name \
    --region=region \
    --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_SCALA_VERSION-CONNECTOR_VERSION.jar
```
  Jar 2.12 kompatibel dengan cluster Dataproc yang dibuat dengan gambar versi 1.5 atau yang lebih baru. Jika cluster Dataproc Anda dibuat dengan Gambar 1.3 atau 1.4, tentukan jar 2.11 sebagai gantinya. Lihat Menyediakan konektor untuk aplikasi Anda untuk informasi selengkapnya.

Output regresi linear (ringkasan model) muncul di jendela terminal ketika tugas selesai.

&lt;&lt;&lt; # Mencetak ringkasan model.
... cetak "Koefisien:" + str(model.coefficients)
Coefficients:[0.0166657454602,-0.00296751984046,0.235714392936,0.00213002070133,-0.00048577251587]
&lt;&lt;&lt; cetak "Mencegat:" + str(model.intersepsi)
Mencegat:-2.26130330748
&lt;&lt;&lt; cetak "R^2" + str(model.summary.r2)
R^2:0,295200579035
&lt;&lt;&lt; model.summary.residuals.show()
+--------------------+
| residual|
+--------------------+
| -0,7234737533344147|
| -0,985466980630501|
| -0,6669710598385468|
| 1,4162434829714794|
|-0,09373154375186754|
|-0,15461747949235072|
| 0,32659061654192545|
| 1,5053877697929803|
| -0,640142797263989|
| 1,229530260294963|
|-0,03776160295256...|
| -0,5160734239126814|
| -1,5165972740062887|
| 1,3269085258245008|
| 1,7604670124710626|
| 1,2348130901905972|
| 2,318660276655887|
| 1,0936947030883175|
| 1,0169768511417363|
| -1,7744915698181583|
+--------------------+
yang hanya menampilkan 20 baris teratas.

Pembersihan

Setelah menyelesaikan tutorial, Anda dapat membersihkan resource yang dibuat agar resource tersebut berhenti menggunakan kuota dan dikenai biaya. Bagian berikut menjelaskan cara menghapus atau menonaktifkan resource ini.

Menghapus project

Cara termudah untuk menghilangkan penagihan adalah dengan menghapus project yang Anda buat untuk tutorial.

Untuk menghapus project:

Perhatian: Menghapus project memiliki efek berikut:

Semua hal dalam project akan dihapus. Jika menggunakan project yang sudah ada untuk tugas dalam dokumen ini, saat Anda menghapusnya, pekerjaan lain yang telah Anda lakukan dalam project tersebut juga akan terhapus.
Project ID kustom hilang. Saat membuat project ini, Anda mungkin telah membuat project ID kustom yang ingin digunakan di masa mendatang. Untuk mempertahankan URL yang menggunakan project ID, seperti URL appspot.com, hapus resource yang dipilih di dalam project, bukan menghapus seluruh project.

Jika Anda berencana mempelajari berbagai arsitektur, tutorial, atau panduan memulai, menggunakan kembali project dapat membantu agar Anda tidak melampaui batas kuota project.

Di konsol Google Cloud, buka halaman Manage resource.
Buka Manage resource
Pada daftar project, pilih project yang ingin Anda hapus, lalu klik Delete.
Pada dialog, ketik project ID, lalu klik Shut down untuk menghapus project.

Menghapus cluster Dataproc

Baca artikel Menghapus cluster.

Langkah selanjutnya

Lihat Tips penyesuaian tugas Spark

Menggunakan Dataproc, BigQuery, dan Apache Spark ML untuk Machine Learning

Tujuan

Biaya

Sebelum memulai

Membuat Subkumpulan data natality BigQuery

Konsol

Python

Menjalankan regresi linear

Konsol

gcloud

Pembersihan

Menghapus project

Menghapus cluster Dataproc

Langkah selanjutnya

Membuat Subkumpulan data `natality` BigQuery