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


Konektor BigQuery untuk Apache Spark memungkinkan Data Scientist memadukan kecanggihan mesin SQL BigQuery yang skalabel tanpa hambatan dengan kemampuan Machine Learning Apache Spark. 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 membuat 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, yang ditulis ke project Google Cloud
  • 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 sudah terinstal. Untuk menyiapkan cluster Dataproc dan menjalankan kode dalam contoh ini, Anda harus melakukan (atau telah melakukan) hal berikut:

  1. 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.
  2. Di konsol Google Cloud, pada halaman pemilih project, pilih atau buat project Google Cloud.

    Buka pemilih project

  3. Enable the Dataproc, BigQuery, Compute Engine APIs.

    Enable the APIs

  4. Menginstal Google Cloud CLI.
  5. Untuk initialize gcloud CLI, jalankan perintah berikut:

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

    Buka pemilih project

  7. Enable the Dataproc, BigQuery, Compute Engine APIs.

    Enable the APIs

  8. Menginstal Google Cloud CLI.
  9. Untuk initialize gcloud CLI, jalankan perintah berikut:

    gcloud init
  10. Buat cluster Dataproc di project Anda. Cluster Anda harus menjalankan versi Dataproc dengan Spark 2.0 atau yang lebih tinggi, (termasuk library machine learning).

Buat Subset data natality BigQuery

Di bagian ini, Anda akan membuat set data di project, lalu membuat tabel dalam set data tempat Anda menyalin subset data tingkat kelahiran dari set data BigQuery natality yang tersedia secara publik. Nantinya di tutorial ini, Anda akan menggunakan data subset dalam tabel ini untuk memprediksi berat lahir sebagai fungsi usia ibu, usia ayah, dan minggu kehamilan.

Anda dapat membuat subkumpulan data menggunakan Konsol Google Cloud atau menjalankan skrip Python di komputer lokal.

Konsol

  1. Buat set data di project Anda.

    1. Buka UI Web BigQuery.
    2. Di panel navigasi kiri, klik nama project Anda, lalu klik CREATE DATASET.
    3. Dalam dialog Create dataset:
      1. Untuk ID Set Data, 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 Akhir masa berlaku tabel default, pilih salah satu opsi berikut:
        • Tidak pernah (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:
      5. Klik Create dataset.
  2. Jalankan kueri terhadap set data kelahiran publik, lalu simpan hasil kueri tersebut dalam tabel baru di set data Anda.

    1. Salin dan tempel kueri berikut ke dalam Query Editor, 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
      
    2. Setelah kueri selesai (dalam waktu sekitar satu menit), hasilnya disimpan sebagai tabel BigQuery "regression_input" 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 mengetahui informasi selengkapnya, lihat dokumentasi referensi API Python Dataproc.

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

  1. Lihat Menyiapkan Lingkungan Pengembangan Python untuk mengetahui petunjuk cara menginstal Python dan Library Klien Google Cloud untuk Python (diperlukan untuk menjalankan kode). Sebaiknya instal dan gunakan Python virtualenv.

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

    """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
  3. 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

  1. Salin dan tempel kode berikut ke file natality_sparkml.py baru 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()
    
    
    

  2. Salin file natality_sparkml.py lokal ke bucket Cloud Storage di project Anda.

    gsutil cp natality_sparkml.py gs://bucket-name
    

  3. Jalankan regresi dari halaman Mengirim tugas Dataproc.

    1. Di kolom Main python file, masukkan URI gs:// bucket Cloud Storage tempat salinan file natality_sparkml.py Anda berada.

    2. Pilih PySpark sebagai Job type.

    3. Masukkan gs://spark-lib/bigquery/spark-bigquery-latest_2.12.jar di kolom File Jar. Hal ini membuat spark-bigquery-connector tersedia untuk aplikasi PySpark saat runtime agar aplikasi dapat membaca data BigQuery ke dalam Spark DataFrame.

    4. Isi kolom Job ID, Region, dan Cluster.

    5. Klik Submit untuk menjalankan tugas di cluster.

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

gcloud

  1. Salin dan tempel kode berikut ke file natality_sparkml.py baru 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()
    
    
    

  2. Salin file natality_sparkml.py lokal ke bucket Cloud Storage di project Anda.

    gsutil cp natality_sparkml.py gs://bucket-name
    

  3. Kirim tugas Pyspark ke layanan Dataproc dengan menjalankan perintah gcloud, yang ditunjukkan di bawah ini, dari jendela terminal di komputer lokal Anda.

    1. Nilai flag --jars membuat spark-bigquery-connector tersedia untuk jobv PySpark saat runtime agar dapat membaca 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
      

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

<<< # Cetak ringkasan model.
30.60-15.60-15.40-15-15.15-15-15_15-15-15.15-15.15-15_15-15-15.15-15-15-15.5-15-15-15_15_15_15-15_15_15-15_15-15-15-15

  

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:

  1. Di konsol Google Cloud, buka halaman Manage resource.

    Buka Manage resource

  2. Pada daftar project, pilih project yang ingin Anda hapus, lalu klik Delete.
  3. Pada dialog, ketik project ID, lalu klik Shut down untuk menghapus project.

Menghapus cluster Dataproc

Lihat Menghapus cluster.

Langkah selanjutnya