機械学習で Dataproc、BigQuery、Apache Spark ML を使用する

Apache Spark 用の BigQuery コネクタを使用すると、データ サイエンティストは、BigQuery のシームレスでスケーラブルな SQL エンジンの能力と Apache Spark の機械学習機能を融合できます。このチュートリアルでは、Dataproc、BigQuery、Apache Spark ML を使用してデータセットで機械学習を実施する方法を示します。

目標

線形回帰を使用して次の 5 つの要素に基づく出生時体重のモデルを構築します。

  1. 妊娠週
  2. 母親の年齢
  3. 父親の年齢
  4. 母親の妊娠中の体重増加
  5. アプガースコア

Google Cloud Platform プロジェクトに書き込まれる線形回帰入力テーブルの作成には、BigQuery が使用されます。BigQuery でのデータのクエリと管理には、Python を使用します。結果の線形回帰テーブルへのアクセスには Apache Spark が使用され、モデルの構築と評価には Spark ML が使用されます。Spark ML 関数の呼び出しには、Dataproc PySpark ジョブが使用されます。

料金

このチュートリアルでは、以下を含む、Google Cloud Platform の課金対象となるコンポーネントを使用します。

  • Compute Engine
  • Dataproc
  • BigQuery

料金計算ツールを使うと、予想使用量に基づいて費用の見積もりを出すことができます。Cloud Platform を初めて使用する方は、無料トライアルをご利用いただけます。

始める前に

Dataproc クラスタには、Spark ML などの Spark コンポーネントがインストールされています。Dataproc クラスタをセットアップしてこの例で示すコードを実行するには、次のことを行う(またはすでに行っている)必要があります。

  1. Google アカウントにログインします。

    Google アカウントをまだお持ちでない場合は、新しいアカウントを登録します。

  2. Cloud Console のプロジェクト セレクタページで、Cloud プロジェクトを選択または作成します。

    プロジェクト セレクタのページに移動

  3. Dataproc, BigQuery, Compute Engine API を有効にします。

    API を有効にする

  4. Cloud SDK をインストールして初期化します。
  5. プロジェクト内で Dataproc クラスタを作成します。クラスタでは、Spark 2.0 以降の Dataproc バージョン(機械学習ライブラリが含まれているバージョン)が実行されている必要があります。

BigQuery natality データのサブセットを作成する

このセクションでは、プロジェクトにデータセットを作成し、そのデータセットにテーブルを作成して、公表されている出生率 BigQuery データセットから出生率データのサブセットをコピーします。このチュートリアルの後半では、このテーブルのサブセット データを使用して、母親の年齢、父親の年齢、妊娠週に基づいて出生体重を予測します。

データ サブセットは、Google Cloud Console を使用するか、ローカルマシンで Python スクリプトを実行して作成できます。

Console

  1. プロジェクトにデータセットを作成します。

    1. BigQuery ウェブ UI に移動します
    2. 左側のナビゲーション ペインでプロジェクト名をクリックし、続いて [データセットを作成] をクリックします。
    3. [データセットを作成] ダイアログで、次のように指定します。
      1. [データセット ID] に「natality_regression」と入力します。
      2. [データのロケーション] では、データセットのロケーションを選択できます。ロケーションのデフォルト値は、US multi-region です。データセットの作成後はロケーションを変更できません。
      3. [デフォルトのテーブルの有効期限] では、次のいずれかのオプションを選択します。
        • [無期限](デフォルト): テーブルは手動で削除する必要があります。
        • [日数]: テーブルは、作成日から指定した日数後に削除されます。
      4. [暗号化] では、次のいずれかのオプションを選択します。
      5. [データセットを作成] をクリックします。
  2. 公表されている出生率データセットに対してクエリを実行し、クエリ結果をデータセット内の新しいテーブルに保存します。

    1. 次のクエリをクエリエディタにコピーして貼り付け、[実行] をクリックします。
      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. 約 1 分後にクエリが完了したら、[結果の保存] をクリックし、続いて保存オプションを選択して、結果を「regression_input」BigQuery テーブルとしてプロジェクトの natality_regression データセットに保存します。

Python

  1. Python および Python 用 Google Cloud クライアント ライブラリ(コードの実行に必要)をインストールする方法については、Python 開発環境の設定をご覧ください。Python virtualenv をインストールして使用することをおすすめします。

  2. 下の natality_tutorial.py コードをローカルマシンの python シェルにコピーして貼り付けます。シェルで <return> キーを押してコードを実行し、デフォルトの Google Cloud プロジェクトに「natality_regression」BigQuery データセットを作成して、公表されている natality データのサブセットを「regression_input」テーブルに設定します。

    """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 = bigquery.Dataset(client.dataset(dataset_id))
    
    # Create the new BigQuery dataset.
    dataset = client.create_dataset(dataset)
    
    # In the new BigQuery dataset, create a reference to a new table for
    # storing the query results.
    table_ref = dataset.table("regression_input")
    
    # Configure the query job.
    job_config = bigquery.QueryJobConfig()
    
    # Set the destination table to the table reference created above.
    job_config.destination = table_ref
    
    # 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.
    query_job = client.query(query, job_config=job_config)
    query_job.result()  # Waits for the query to finish
  3. natality_regression データセットと regression_input テーブルの作成を確認します。

線形回帰を実行する

このセクションでは、Google Cloud Console を使用して Dataproc サービスにジョブを送信するか、ローカルのターミナルから gcloud コマンドを実行して PySpark 線形回帰を実行します。

Console

  1. ローカルマシンの新しい natality_sparkml.py ファイルに、次のコードをコピーして貼り付けます。

    """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 __future__ import print_function
    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. ローカルの natality_sparkml.py ファイルをプロジェクトの Cloud Storage バケットにコピーします。

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

  3. Dataproc の [ジョブを送信] ページから回帰を実行します。

    1. [メインの Python ファイル] フィールドで、natality_sparkml.py ファイルのコピーが置かれている Cloud Storage バケットの gs:// URI を挿入します。

    2. [ジョブタイプ] として PySpark を選択します。

    3. gs://spark-lib/bigquery/spark-bigquery-latest.jarJAR ファイル フィールドに挿入します。これにより、実行時に spark-bigquery-connector を PySpark アプリケーションで使用できるようになり、BigQuery データを Spark DataFrame に読み込めます。

    4. [ジョブ ID]、[リージョン]、[クラスタ] フィールドに入力します。

    5. [送信] をクリックして、クラスタでジョブを実行します。

ジョブが完了すると、線形回帰出力モデルの概要が Dataproc の [ジョブの詳細] ウィンドウに表示されます。

gcloud コマンド

  1. ローカルマシンの新しい natality_sparkml.py ファイルに、次のコードをコピーして貼り付けます。

    """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 __future__ import print_function
    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. ローカルの natality_sparkml.py ファイルをプロジェクトの Cloud Storage バケットにコピーします。

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

  3. 下に示すようにローカルマシンのターミナル ウィンドウから gcloud コマンドを実行することにより、Pyspark ジョブを Dataproc サービスに送信します。

    1. --files フラグの値には、natality_sparkml.py ファイルのコピーがある Cloud Storage バケットの名前を入れます。
    2. --jars フラグの値を指定することで、実行時に spark-bigquery-connector を PySpark jobv で使用できるようになり、BigQuery データを Spark DataFrame に読み込めます。
      gcloud dataproc jobs submit pyspark \
          --cluster=cluster-name \
          --region=region \
          --files=gs://your-bucket/natality_sparkml.py \
          --jars=gs://spark-lib/bigquery/spark-bigquery-latest.jar
      

ジョブが完了すると、線形回帰出力(モデル概要)がターミナル ウィンドウに表示されます。

<<< # Print the model summary.
... print "Coefficients:" + str(model.coefficients)
Coefficients:[0.0166657454602,-0.00296751984046,0.235714392936,0.00213002070133,-0.00048577251587]
<<< print "Intercept:" + str(model.intercept)
Intercept:-2.26130330748
<<< print "R^2:" + str(model.summary.r2)
R^2:0.295200579035
<<< model.summary.residuals.show()
+--------------------+
|           residuals|
+--------------------+
| -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|
+--------------------+
only showing top 20 rows.

クリーンアップ

「Dataproc」、「Biguery」、「Spark ML for Machine Learning」のチュートリアルが終了したら、Google Cloud で作成したリソースをクリーンアップして、今後料金が発生しないようにします。次のセクションで、リソースを削除または無効にする方法を説明します。

プロジェクトの削除

課金をなくす最も簡単な方法は、チュートリアル用に作成したプロジェクトを削除することです。

プロジェクトを削除するには:

  1. Cloud Console で [リソースの管理] ページに移動します。

    [リソースの管理] ページに移動

  2. プロジェクト リストで、削除するプロジェクトを選択し、[削除] をクリックします。
  3. ダイアログでプロジェクト ID を入力し、[シャットダウン] をクリックしてプロジェクトを削除します。

Dataproc クラスタの削除

クラスタの削除をご覧ください。