Databricks を活用した データ分析によるビジネス価値の創出
によって Takaaki Yayoi による投稿
Community Editionに代わり、Free Editionでは無料でより充実した機能をご利用いただけます。ぜひ今日からぜひFree Editionをお試しください。
こちらはデータブリックスユーザー会の第一回イベントで発表された内容となります。こちらからサンプルノートブックをダウンロードできます。
この記事では、データ分析を通じでビジネス価値創出につなげるために辿るプロセスを、実例を含めてご紹介します。
個人的経験を踏まえたものですが、データ分析プロジェクトは以下のフローになるかと思います。最後までたどり着けないプロジェクトもたくさんありました。
いきなりデータ分析を始めるプロジェクトはまず存在しません。データ分析には必ずビジネスにつながる目的があるべきです。
ちなみに意外と大変なのはステップ 4 です。データの由来、スキーマ、更新頻度などの確認です。いろんな人に聞いて回らないとわからないケースもしばしばです。
上で述べたとおり、デモグラフィック情報に基づいて、年収が 5 万ドル以上か否かを分類する二値分類問題に取り組みます。データセットは、UCI Machine Learning Repository にあるものを利用します。このデータは既に Databricks ランタイムに格納されています。本記事では、データ処理、機械学習パイプライン、機械学習アルゴリズムなどの MLlib の機能をデモンストレーションします。
大抵の場合、分析データを入手した後にすることはデータの中身の確認でしょう。Databricks では柔軟にノートブック上での作業を行えるように、多くのマジックコマンドがサポートされています。下のセルにある %fs もその 1 つです。Databricks のファイルシステムに格納されているファイルの一部を表示します。
| %fs head –maxBytes=1024 databricks-datasets/adult/adult.data |
データセットにはカラム名が含まれていないため、カラム名とデータタイプを指定するスキ ーマを作成します。作成したスキーマを指定して CSV ファイルを読み込みます。
# スキーマの定義
# ファイルを読み込みます |
データを理解するための EDA は非常に重要です。EDA を通じて取り扱うデータの素性を理解することで、以降のデータ分析での手戻りを減らすことができます。
モデル構築、評価に向けて、データセットをランダムにトレーニングデータとテストデータに分割します。また、再現性を確保するために乱数のシードを設定しています。
あらゆる前処理を実行する前の生の状態でデータを分割すべきです。これにより、モデルを評価する際、テストデータが未知のデータに近い状態を維持することができます。
| trainDF, testDF = dataset.randomSplit([0.8, 0.2], seed=42) print(“トレーニングデータ:”, trainDF.cache().count()) # 何回かトレーニングするのでデータをキャッシュします print(“テストデータ:”, testDF.count()) |
デー タを確認しましょう。EDA! EDA!
| # データを確認するには、とにかく display! display(trainDF) |
| display(trainDF.select(“hours_per_week”).summary()) |
education（最終学歴）はどうなっているでしょうか？
とは言っても、表形式で全体傾向を把握するのには限界があります。「百のテーブルは一のグラフにしかず」です。そんな時には Let’s 可視化！をクリック !
| # 最終学歴でグルーピングして件数をカウント、カウントの昇順でソートして表示 display(trainDF .groupBy(“education”) .count() .sort(“count”, ascending=False)) |
本記事において、機械学習モデルを構築する際には Apache Spark の機械学習ライブラリ MLlib を使用します。MLlib の機械学習における重要な 3 つのコンセプトは、Transformers、Estimators、そして Pipelines です。
Pipeline の詳細に関してはこちらを参照ください。
ここでのゴールは、データセットに含まれる特徴量（教育レベル、既婚・未婚、職業など）から、年収 income のレベルを予測するというものです。まず、MLlib で利用できるように特徴量を操作、前処理を行います。いわゆる特徴量エンジニアリングです。
線形回帰、ロジスティック回帰などの学習アルゴリズムでは、特徴量が数値である必要があります。上記の成人データセットでは、教育、職業、既婚・未婚のデータがカテゴリ変数となっています。
以下のコードでは、カテゴリ変数を 0 か 1 のみを取る数値変数に変換するために、どのように StringIndexer と OneHotEncoder を使用す るのかを説明します。
Sparkにおける One-hot エンコーディングは 2 段階のプロセスとなります。最初にStringIndexer を使い、OneHotEncoder を呼び出します。以下のコードブロックでは、StringIndexer と OneHotEncoder を定義しますが、まだデータには適用しません。
from pyspark.ml.feature import StringIndexer, OneHotEncoder
categoricalCols = [“workclass”, “education”, “marital_status”, “occupation”, “relationship”, “race”, “sex”] # 以下の 2 行は estimator となります。後ほどデータセットを変換する際に適用することになる関数を返却します。 # ラベルとなるカラム (“income”) も (“<=50K"、">50K”) の 2 つの値をとる文字列のカラムとなります。 |
ここでは、特徴量エンジニアリングとモデル構築のステップ全てを 1 つのパイプラインにまとめます。その前に上のコードブロックで構築した stringIndexer estimator を適用することで estimator や transformer がどのように動作するのかを詳しく見てみましょう。
データセットを変換する StringIndexerModel を返却するように .fit() メソッドを呼び出します。
そして、StringIndexerModel の .transform() メソッドを呼び出すことで、特徴量を変換し、結果を格納するカラム …Index が追加された新たなデータフレームが返却されます。必要であれば、表示結果を右にスクロールして追加されたカラムを参照してください。
| stringIndexerModel = stringIndexer.fit(trainDF) display(stringIndexerModel.transform(trainDF)) |
多くの MLlib アルゴリズムでは、入力として単一の特徴量カラムが必要となります。それぞれの行の特徴量カラムは、予測に用いる特徴量に対応するベクトルを保持します。
MLlibは、一連のカラムから単一のベクトルカラムを作成する VectorAssembler transformer を提供します。
下のコードブロックではどのように VectorAssembler を使用するのかを説明します。
from pyspark.ml.feature import VectorAssembler
# ここには、データセットの数値カラムと one-hot エンコードされた2値のベクトル両方が含まれます。 |
分析アルゴリズムは、データとビジネス課題に応じて使い分けることになると思います。EDA の過程でも「この辺りのアルゴリズムだろうな」などと当たりをつけながらデータを見ていきます。
どのタイミングでレビューを行うのかは、チームやプロジェクトによるのではないでしょうか。しかし、リモートワークが浸透した昨今、ここまで開発したロジックを簡単にシェアできる機能があると便利だと思いませんか？
安心してください。Databricks にはあります！
以下の機能を利用いただくことで、チーム間で連携しながら開発を進めることができます。
ここでは予測にロジスティック回帰（英語）モデルを使います。
Pipeline は、transformers、estimators が含まれる順番付きのリストです。データセットに適用する変換処理の再現性を確保し、自動化するために、パイプラインを定義することができます。
StringIndexer で見たのと同様に、Pipeline も estimator です。pipeline.fit() メソッドが、transformer である PipelineModel を返却します。
from pyspark.ml import Pipeline
# これまでに作成したステージを組み合わせてパイプラインを定義します # パイプラインモデルを定義します # テストデータセットにパイプラインモデルを適用します |
モデルによる予測結果を表示します。features カラムは、one-hot エンコーディングを実行した後、多くのケースで要素のほとんどが 0 となる sparse vector（英語）となります。
| display(predDF.select(“features”, “label”, “prediction”, “probability”)) |
モデルの評価方法にもいろいろありますが、ここでは ROC カーブで評価を行います。ここまで使用してきている display コマンドには、ROC カーブを表示するオプションが組み込まれています。
| display(pipelineModel.stages[-1], predDF.drop(“prediction”, “rawPrediction”, “probability”), “ROC”) |
モデル評価に必要な ROC カーブの AUC（Area Under the Curve）を計算するために BinaryClassificationEvaluator を用い、精度を評価するために MulticlassClassificationEvaluator を用います。
参考情報：
from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator
bcEvaluator = BinaryClassificationEvaluator(metricName=”areaUnderROC”) mcEvaluator = MulticlassClassificationEvaluator(metricName=”accuracy”) |
AUC が 0.88 となっており、それなりの精度であることがわかります。
一度のトレーニングで得られたモデルをベストモデルと考えるのは早計です。通常はハイパラメータをチューニングしながらモデルの改善を行っていくことになります。MLlib はハイパーパラメータチューニングと交差検証（cross validation）の手段を提供します。
モデルをチューニングするために、ParamGridBuilder と CrossValidator を使用します。本例においては、CrossValidator での検証において、3 種類の regParam、3 種類の elasticNetParam から生成される、3 x 3 = 9 のハイパーパラメータの組み合わせを使用します。
from pyspark.ml.tuning import ParamGridBuilder, CrossValidator
paramGrid = (ParamGridBuilder() |
MLlibの CrossValidator を呼び出した際、Databricks は MLflow を用いて、自動的に全てのラン（モデルのトレーニング）を追跡します。MLflow の UI（AWS | Azure）を用いて、構築したモデルを比較することができます。下のセルの実行後、画面右上にある Experiment ボタンを押してみてください。MLflow に関してはこちらの記事を参照ください。
本例では、作成したパイプラインを estimator とします。
# 3フォールドのCrossValidatorを作成
import mlflow with mlflow.start_run(): # ベストモデルをテストデータで評価しロギングします。 # ベストモデルをロギングします。 |
ベストモデルをデプロイ（配備）して、予測を実行します。デプロイには、いくつかの方法があります。
MLflow を活用することで、モデルサービングおよびモデルの本格デプロイに向けた承認フローなどを容易に構築することができます。
ここでは、テストデータセットに対する予測を行うために、交差検証によって特定されたベストモデルを用います。
| # テストデータセットに対す る予測を行うために、交差検証によって特定されたベストモデルを使用 cvPredDF = cvModel.transform(testDF) |
予測結果のデータセットを見てみます。prediction カラムの値が 0 の場合、<=50K、1 の場合 >50K と予測したことを意味します。
| display(cvPredDF) |
また、SQL を用いて予測結果を年齢別、職業別に集計することができます。SQL を実行するためには、予測結果のデータセットから一時ビューを作成します。
| cvPredDF.createOrReplaceTempView(“finalPredictions”) |
職業ごとの予測結果を見てみます。以下のグラフの prediction は以下のとおり解釈します。
| %sql SELECT occupation, prediction, count(*) AS count FROM finalPredictions GROUP BY occupation, prediction ORDER BY occupation |
経営者、専門職などでは、高い割合で高収入と予測する傾向が見て取れます。
次に、予測結果を年齢ごとに見てみます。
| %sql SELECT age, prediction, count(*) AS count FROM finalPredictions GROUP BY age, prediction ORDER BY age |
実運用においては、モデルが定常的に目標とする精度を達成しているのかをモニタリングする必要があります。モデルの経年劣化「ドリフト」を検知した場合には、再学習を行うなどの対応が必要となります。
なお、ドリフトには以下の種類があります。
概念ドリフト（concept drift）
目標変数の統計的属性が変化した時、予測しようとする本当の概念もまた変化します。例えば、不正トランザクションにおいては、新たな手口が生まれてくると、不正の定義自体を見直さなくてはなりません。このような変化は概念ドリフトを引き起こします。
データドリフト（data drift）
入力データから選択された特徴量を用いてモデルをトレーニングします。入力データの統計的特性に変化が生じた際に、モデルの品質に影響を及ぼします。例えば、季節性によるデータの変化、個人的嗜好の変化、トレンドなどは入力データのドリフトを引き起こします。
上流データの変化（upstream data changes）
モデル品質に影響を与えうるデータパイプライン上流でのオペレーションの変更が生じる場合があります。例えば、特徴量のエンコーディングにおいて華氏から摂氏に変更があった場合や、特徴量の生成が停止されることで null や欠損値になるなどです。
ここでは、ベストモデルを AUC で評価しますが、実際には評価は一度限りではなく、継続的なオペレーションになります。これがデータ分析が 「終わり無き」プロセスたる所以です。
モデルドリフトに関してはこちらの記事を参照ください。
| # AUCと精度を用いてモデルの性能を評価 print(f”Area under ROC curve: {bcEvaluator.evaluate(cvPredDF)}”) print(f”Accuracy: {mcEvaluator.evaluate(cvPredDF)}”) |
今回は、データ分析の終わり無きプロセスの一部フローをお見せしました。
ほぼ全ての作業を Databricks のワークスペースで完結することができることをご覧いただけたかと思います。
是非、Databricks をご利用ください！
ブログを購読して、最新の投稿を受信トレイにお届けします。