データ分析

新しい初期化アクションで Dataproc の機械学習を高速化

2020年12月23日

Google Cloud Japan Team

※この投稿は米国時間 2020 年 12 月 12 日に、Google Cloud blog に投稿されたものの抄訳です。

Apache Hadoop と Apache Spark は、分散ストレージとデータ処理のための確立された標準フレームワークです。Google Cloud のフルマネージドクラウドサービスである Dataproc を Apache Spark および Apache Hadoop クラスタの実行に使用することで、データサイエンティストとデータエンジニアがこれらのフレームワークを簡単に使用して、コストを節約できます。機械学習モデルを構築する場合は、Apache Spark を使用したデータの前処理に Dataproc を使用できます。また、同じ Spark クラスタを使用して、ノートブックを機械学習用に強化できます。Google は機械学習用クラスタを構成する時間を節約するために、機械学習の初期化アクションを作成して、よく使用されるライブラリセットを提供しています。

このブログでは、Dataproc クラスタでこの初期化アクションの使用を開始する方法をご紹介します。これにより、以下のような最新かつ最高水準のオープンソースの機械学習を活用するための環境を整えることができます。

TensorFlow、PyTorch、MxNet、Scikit Learn、Keras などの Python パッケージ
XGBoost、Caret、randomForest、sparklyr などの R パッケージ
Spark-BigQuery コネクタ
Dask と Dask-Yarn
Spark 上のRAPIDS（必要に応じて）
GPU とドライバ（必要に応じて）

さらに、オプションコンポーネントとコンポーネントゲートウェイを通じて Jupyter や Zeppelin を使い、Dataproc のエクスペリエンスを向上できます。

機械学習の初期化アクションは、RAPIDS、Dask、GPU ドライバなど特定のコンポーネントをインストールするために、他の初期化アクションを使用します。そのため、ユーザーはこれらのコンポーネントのすべての機能と構成にアクセスできます。

データの前処理と機械学習トレーニングを 1 か所で

Dataproc の機械学習の初期化アクションは、ユーザーのデータで Spark、Dask などの ETL ジョブを実行する本番環境を提供します。同時に、同じ場所で選択した機械学習ライブラリを使用して機械学習モデルをビルドできます。クラスタ構成に GPU を追加することで、TensorFlow や RAPIDS で機械学習モデルのトレーニングにかかる時間を短縮できます。また、RAPIDS SQL Accelerator を使用して、モデルトレーニングの効率をさらに改善できます。

Google Cloud Platform プロジェクトの構成

Google Cloud Platform（GCP）プロジェクトが必要です。既存のプロジェクトを使用するか、こちらの手順に沿って新しいプロジェクトを作成してください。

機械学習の初期化アクションは、一部のインストールに他の初期化アクションを使用します。以下のスクリプトのコピーを作成して、使用中のスクリプトのバージョンを効果的に「固定」できます。まず、Cloud Storage バケットを作成し、作成したバケットにスクリプトをコピーします。

読み込んでいます...

機械学習の初期化アクションを使用して Dataproc クラスタを作成する

クラスタを作成するには、まずリージョンとクラスタ名を定義します。

読み込んでいます...

次に、以下のコマンドを実行すると、機械学習の初期化アクションにより構成されたクラスタが作成されます。クラスタには、マスターノードが 1 つ、ワーカーノードが 2 つ、各ノードで利用できる NVIDIA T4 GPU が 2 つ含まれます。また、クラスタのオプション Jupyter コンポーネントがコンポーネントゲートウェイとともに有効化され、Jupyter Notebook か JupyterLab を使用してクラスタにアクセスできるようになります。

読み込んでいます...

上の構成は、NVIDIA のグラフィックスカードを搭載し、それぞれのドライバがインストールされた Dataproc クラスタを作成するものです。これにより、NVIDIA RAPIDS や TensorFlow などのフレームワークを使用して、データ処理を GPU で高速化できるようになります。

この構成に「init-actions-repo」メタデータフラグを含めることで、機械学習の初期化アクションに必要なインストールスクリプトの場所を示すことができます。さらに、「include-gpus=true」と「gpu-driver-provider=NVIDIA」のフラグはそれぞれ、スクリプトに GPU ドライバをインストールする必要があること、ドライバが NVIDIA から提供されていることを示します。必要に応じて、GPU やドライバなしでクラスタを実行できます。

また、NVIDIA RAPIDS、Spark の JAR、Dask 用 NVIDIA RAPIDS を使用することも可能です。その場合、「rapids-runtime」メタデータフラグを使用して、これを DASK または RAPIDS に割り当てます。

spark-tensorflow-distributor を使用して分散 TensorFlow ジョブを実行する

機械学習の初期化アクションに含まれる spark-tensorflow-distributor を使用して、Spark クラスタで分散 TensorFlow ジョブを実行できます。このライブラリは、TensorFlow 分散ライブラリのラッパーです。以下のコードを「spark_tf_dist.py」ファイルにコピーします。

読み込んでいます...

from pyspark.sql import SparkSession
from spark_tensorflow_distributor import MirroredStrategyRunner

spark = (SparkSession.builder.appName("tf")
  .config("spark.dynamicAllocation.enabled", "false")
  .config("spark.executor.resource.gpu.amount", "1")
  .config("spark.executor.cores", "15")
  .config("spark.task.cpus", "15")
  .config("spark.task.resource.gpu.amount", "1")
  .config("spark.executor.resource.gpu.discoveryScript", "/usr/lib/spark/scripts/gpu/getGpusResources.sh")
  .config("spark.driver.resource.gpu.discoveryScript", "/usr/lib/spark/scripts/gpu/getGpusResources.sh")
  .config("spark.driver.resource.gpu.amount", "1")         
  .getOrCreate())

def train():
    import tensorflow as tf
    import uuid

BUFFER_SIZE = 10000
    BATCH_SIZE = 64

def make_datasets():
        (mnist_images, mnist_labels), _ = \
            tf.keras.datasets.mnist.load_data(path=str(uuid.uuid4())+'mnist.npz')

dataset = tf.data.Dataset.from_tensor_slices((
            tf.cast(mnist_images[..., tf.newaxis] / 255.0, tf.float32),
            tf.cast(mnist_labels, tf.int64))
        )
        dataset = dataset.repeat().shuffle(BUFFER_SIZE).batch(BATCH_SIZE)
        return dataset

def build_and_compile_cnn_model():
        model = tf.keras.Sequential([
            tf.keras.layers.Conv2D(32, 3, activation='relu', input_shape=(28, 28, 1)),
            tf.keras.layers.MaxPooling2D(),
            tf.keras.layers.Flatten(),
            tf.keras.layers.Dense(64, activation='relu'),
            tf.keras.layers.Dense(10, activation='softmax'),
        ])
        model.compile(
            loss=tf.keras.losses.sparse_categorical_crossentropy,
            optimizer=tf.keras.optimizers.SGD(learning_rate=0.001),
            metrics=['accuracy'],
        )
        return model

train_datasets = make_datasets()
    options = tf.data.Options()
    options.experimental_distribute.auto_shard_policy = tf.data.experimental.AutoShardPolicy.DATA
    train_datasets = train_datasets.with_options(options)
    multi_worker_model = build_and_compile_cnn_model()
    multi_worker_model.fit(x=train_datasets, epochs=3, steps_per_epoch=5)
    
MirroredStrategyRunner(num_slots=2).run(train)

このコードは、MirroredStrategyRunner を使用して、TensorFlow トレーニングジョブを Spark クラスタに送信するものです。Spark 構成が提供されると、クラスタ上でトレーニングに GPU を最大限活用できるようになります。

次に、以下のコードで TensorFlow コードを PySpark ジョブとしてクラスタに送信できます。

読み込んでいます...

Dataproc Hub

機械学習の初期化アクションは、ノートブック環境での使用に適しています。これを行う一つの方法として、Dataproc でオプション Jupyter コンポーネントを使用する方法が挙げられます。こちらのブログ投稿で詳しい情報をご覧ください。また、Dataproc のマネージド JupyterLab サービスである Dataproc Hub を使用するのも良い方法です。IT 管理者はこのサービスを利用して、クラスタに含まれるパッケージとライブラリをカスタマイズできる柔軟性を確保しつつ、セキュリティの確保とリソースの割り振りに最適な環境を事前構成して、データサイエンティストに提供できます。以下の YAML 構成を使用して、機械学習の初期化アクションを使用したクラスタの構成が可能です。詳しい手順はこちらからご覧ください。

読み込んでいます...

Dataproc Hub について詳しくは、こちらのブログ投稿でのお知らせをご覧ください。

次のステップ

機械学習の初期化アクションは、ETL 処理ジョブの実行と機械学習モデルのトレーニングの両方を行うのに最適です。また、オープンソースの初期化アクションを使用すると、方法をカスタマイズできます。さらに、カスタム Dataproc イメージを作成して、クラスタを簡単に作成し、作成時間を短縮できます。機械学習の初期化アクションの利用開始について詳しくは、こちらのドキュメントをご覧ください。Google Cloud Platform で新規のお客様に発行される $300 のクレジットを利用して、Dataproc で Dask を使い始めることもできます。

-デベロッパープログラムエンジニア Brad Miro

投稿先