「BigQuery ML」：SQLで機械学習ってどういうこと？試しにSQLでロジスティック回帰を書いてみた。

はじめに

Google Cloud Next 2018でBigQuery MLが発表されました。

文字通り、「BigQuery + 機械学習(Machine Learning)」を実現するもののようです。さっそく試してみたので、どんなものなのかと、その原理（の一部）を紹介したいと思います。この記事ではBigQuery MLの紹介と、それを直接SQLでやったらどのくらい大変かを見てみます。

BigQueryとは？

BigQuery は、Google が提供するサーバーレスでスケーラビリティに優れた、低コストのエンタープライズ向けデータウェアハウスです

とあるように、Googleのクラウドサービスの一つで、大規模なデータを貯めておくことができ、大量のデータに対しても分散して高速なデータの取得が可能です。また比較的安価な料金で利用することができます。

ちなみん、Wantedlyの中でもBigQueryが使われています。ログの集計や定期的なデータ加工処理、期間をまたいだデータの分析など、データ量が大きいときにBigQueryが利用されています。

BigQuery MLとは？

その名の通り、BigQuery上でMachine Learningを実現する機能です。現段階では、「線形回帰モデル（データから数値を推定する）」と「ロジスティック回帰モデル（データからtrue/falseを判定する）」が既に用意されています。今までは、BigQueryからデータをローカル環境などにコピーしてから分析していたものがBigQueryだけ完結します。しかもBigQueryだから大規模データで動きそうです。

BigQuery MLはこんなことを実現すると言っています。

機械学習の煩雑さを軽減：データをローカルファイルや機械学習環境にコピーする手間がない
セキュリティ向上：コピーしないでBigQueryだけで完結するのでデータが漏洩しずらい
機械学習の民主化：機械学習をツールとして誰でも使えるようにする

料金

BigQueryは料金は入力データのサイズと処理したデータのサイズで決まります。具体的な数字は書かれていませんが、BigQuery MLでは入力データより多くのデータの処理がかかると書かれています。正直これを読んでもいくらかかるかわかりませんでした。

Currently, if you use BigQuery on demand, your BigQuery ML charges are based on the data processed by each query. For BigQuery ML queries, the data processed is usually greater than the just the input data for the CREATE MODEL statement.

BigQury MLの使い方

すごくシンプルなサンプルデータを試してみます。

下のようなデータを分類してみましょう。赤と青を分類する問題としてロジスティック回帰で解けそうです。

学習

これだけです。以下のSQLを実行するとtmp.ml_sampleにモデルが出力されます。

内部ではSQLでデータを作っています。{0, 1}が推定する変数(label)です。

#standardSQL
CREATE OR REPLACE MODEL `tmp.ml_sample`
OPTIONS
  (model_type='linear_reg', input_label_cols=['label']) AS
SELECT
      *
FROM
      UNNEST(ARRAY<STRUCT<label INT64, x1 FLOAT64, x2 FLOAT64>> [
      (0, 4.24, 14.92),
      (0, 11.67, 14.79),
      (0, 16.14, 15.51),
      (0, 2.89, 5.33),
      (0, 11.56, 9.93),
      (1, 7.04, 6.81),
      (1, 10.74, 8.03),
      (1, 13.55, 12.18),
      (1, 8.52, 2.26),
      (1, 14.71, 2.85)
      ])

推論

推論時は学習で作られたモデル名を指定して ML.PREDICTを呼び出すだけです。簡単。

SELECT
  *
FROM
  ML.PREDICT(MODEL `tmp.ml_sample`,
    (
SELECT
      *
FROM
      UNNEST(ARRAY<STRUCT<label INT64, x1 FLOAT64, x2 FLOAT64>> [
      (0, 4.24, 14.92),
      (0, 11.67, 14.79),
      (0, 16.14, 15.51),
      (0, 2.89, 5.33),
      (0, 11.56, 9.93),
      (1, 7.04, 6.81),
      (1, 10.74, 8.03),
      (1, 13.55, 12.18),
      (1, 8.52, 2.26),
      (1, 14.71, 2.85)
      ])
    ))

推論結果↓

predicted_labelが10個中8個を正しく推論できています（わざと線形分離不可能なデータにしました）。

これ以外にも、閾値を段階的に変えてRecallやPrecisionを出すための ML.ROC_CURVE や、ML.WEIGHTS 関数が用意されていて、分析するときに行いそうなことは関数一つでできることを目指している感じがあります。

どうやってSQLで機械学習するの？

さて、どうやってこれが実現されているのでしょうか？ Google AI Blogでは、「SQLのシンプルな拡張であり、Gradient Descentアルゴリズムは通常のSQLの操作で表現できる」と書いてあります。本当かどうか確認したくなります。

the core components of gradient descent — an optimization method that is the workhorse of machine learning algorithms — can be implemented using common SQL operations

For example, a gradient vector can be computed using the SUM and GROUP BY operators, and the weights of a model can be updated using an INNER JOIN.↩

実際にSQLでロジスティック回帰を書いてみましょう！

（以下は、ロジスティック回帰の簡単な最適化を実際には使いもしないのにSQLでむりやり書くものです。興味が無かったら読む必要はないです。）

ロジスティック回帰を復習します。復習

簡単のため変数は2変数だけ、正則化は全く考えないことにします。重みWと説明変数のベクトルXからロジスティック回帰の確率値は以下の式で与えられます。

SQLで書いてみましょう。初期値として適当な重みを与えて、ロジスティック回帰の確率値を出力するコードはSQLで以下のように書くことができます。BigQueryではJSのコードで関数定義ができるため意外と簡単に書くことができました。

# Model function
CREATE TEMPORARY FUNCTION lr_prob(x1 FLOAT64, x2 FLOAT64, w0 FLOAT64, w1 FLOAT64, w2 FLOAT64)
      RETURNS FLOAT64
      LANGUAGE js AS """
        return 1 / (1 + Math.exp(-(w0 + w1 * x1 + w2 * x2)));
      """;

# Initialize Prameters
WITH data AS (
  SELECT
    *
  FROM
    UNNEST(ARRAY<STRUCT<label INT64, x1 FLOAT64, x2 FLOAT64>> [
    (0, 4.24, 14.92),
    (0, 11.67, 14.79),
    (0, 16.14, 15.51),
    (0, 2.89, 5.33),
    (0, 11.56, 9.93),
    (1, 7.04, 6.81),
    (1, 10.74, 8.03),
    (1, 13.55, 12.18),
    (1, 8.52, 2.26),
    (1, 14.71, 2.85)
    ])
),
weights AS (
  SELECT
      *
  FROM
    UNNEST(ARRAY<STRUCT<w0 FLOAT64, w1 FLOAT64, w2 FLOAT64>> [
      (1.0, 0.5, 0.5)
    ])
)

SELECT
  *,
  lr_prob(x1, x2, w0, w1, w2) AS prob,
  lr_prob(x1, x2, w0, w1, w2) > 0.5 AS predict
FROM data
JOIN weights On True

↓出力はこのようになります。predictが0.5を超えているので全て1(true)だと判定されています。

重みの更新式をSQLで書く

次にここから重みWを更新していきます。

更新式は以下で与えられます。要するに予測値から真の値を引いたものを入力ベクトルにかけ合わせてSUMを取るものです。

SQLで書いてみましょう。ちなみに学習率は適当に0.05としました。

結果表示のため、初期状態と更新された状態で予測値を出力してUNION ALLして並べて書いています。

# Model and update function
CREATE TEMPORARY FUNCTION lr_prob(x1 FLOAT64, x2 FLOAT64, w0 FLOAT64, w1 FLOAT64, w2 FLOAT64)
      RETURNS FLOAT64
      LANGUAGE js AS """
        return 1 / (1 + Math.exp(-(w0 + w1 * x1 + w2 * x2)));
      """;

CREATE TEMPORARY FUNCTION delta_weight(predict FLOAT64, answer FLOAT64, x1 FLOAT64, x2 FLOAT64)
      RETURNS ARRAY<FLOAT64>
      LANGUAGE js AS """
        var delta = (predict - answer);
        return [delta, delta * x1, delta * x2];
      """;

# Initialize Prameters
WITH data AS (
  SELECT
    *
  FROM
    UNNEST(ARRAY<STRUCT<label INT64, x1 FLOAT64, x2 FLOAT64>> [
    (0, 4.24, 14.92),
    (0, 11.67, 14.79),
    (0, 16.14, 15.51),
    (0, 2.89, 5.33),
    (0, 11.56, 9.93),
    (1, 7.04, 6.81),
    (1, 10.74, 8.03),
    (1, 13.55, 12.18),
    (1, 8.52, 2.26),
    (1, 14.71, 2.85)
    ])
),
weights AS (
  SELECT
      *
  FROM
    UNNEST(ARRAY<STRUCT<w0 FLOAT64, w1 FLOAT64, w2 FLOAT64>> [
      (1.0, 0.5, 0.5)
    ])
),
# Start Learning
# --- Step 1
weights_1 as (
      SELECT
        SUM(w0) as w0,
        SUM(w1) as w1,
        SUM(w2) as w2
      FROM (
            SELECT * FROM weights
            UNION ALL
            SELECT * FROM (
                  SELECT
                    -0.05 * SUM(delta_w0) / (SELECT COUNT(*) FROM data) as w0,
                    -0.05 * SUM(delta_w1) / (SELECT COUNT(*) FROM data) as w1,
                    -0.05 * SUM(delta_w2) / (SELECT COUNT(*) FROM data) as w2
                  FROM (
                  SELECT
                    *,
                    lr_prob(x1, x2, w0, w1, w2) as predict,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(0)] as delta_w0,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(1)] as delta_w1,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(2)] as delta_w2
                  from data
                  join weights on true
                  )
            )
      )
)

SELECT
  0 as step,
  *,
  lr_prob(x1, x2, w0, w1, w2) AS predict
FROM data
JOIN weights On True

UNION ALL

SELECT
  1 as step,
  *,
  lr_prob(x1, x2, w0, w1, w2) AS predict
FROM data
JOIN weights_1 On True

ORDER BY STEP ASC, label ASC

↓出力はこうなります。更新があるため、stepというカラムを導入しました。step=0が初期状態、step=1が更新1回目の予測値です。

predictの値を見ると少し正解に近づいたもののまだtrueに振り分けられています

この調子でstepを重ねていけば最適化ができそうです。

調子に乗って4ステップまでコピペして書いてみました。

# Model and update function
CREATE TEMPORARY FUNCTION lr_prob(x1 FLOAT64, x2 FLOAT64, w0 FLOAT64, w1 FLOAT64, w2 FLOAT64)
      RETURNS FLOAT64
      LANGUAGE js AS """
        return 1 / (1 + Math.exp(-(w0 + w1 * x1 + w2 * x2)));
      """;

CREATE TEMPORARY FUNCTION delta_weight(predict FLOAT64, answer FLOAT64, x1 FLOAT64, x2 FLOAT64)
      RETURNS ARRAY<FLOAT64>
      LANGUAGE js AS """
        var delta = (predict - answer);
        return [delta, delta * x1, delta * x2];
      """;

# Initialize Prameters
WITH data AS (
  SELECT
    *
  FROM
    UNNEST(ARRAY<STRUCT<label INT64, x1 FLOAT64, x2 FLOAT64>> [
    (0, 4.24, 14.92),
    (0, 11.67, 14.79),
    (0, 16.14, 15.51),
    (0, 2.89, 5.33),
    (0, 11.56, 9.93),
    (1, 7.04, 6.81),
    (1, 10.74, 8.03),
    (1, 13.55, 12.18),
    (1, 8.52, 2.26),
    (1, 14.71, 2.85)
    ])
),
weights AS (
  SELECT
      *
  FROM
    UNNEST(ARRAY<STRUCT<w0 FLOAT64, w1 FLOAT64, w2 FLOAT64>> [
      (1.0, 0.5, 0.5)
    ])
),
# Start Learning
# --- Step 1
weights_1 as (
      SELECT
        SUM(w0) as w0,
        SUM(w1) as w1,
        SUM(w2) as w2
      FROM (
            SELECT * FROM weights
            UNION ALL
            SELECT * FROM (
                  SELECT
                    -0.05 * SUM(delta_w0) / (SELECT COUNT(*) FROM data) as w0,
                    -0.05 * SUM(delta_w1) / (SELECT COUNT(*) FROM data) as w1,
                    -0.05 * SUM(delta_w2) / (SELECT COUNT(*) FROM data) as w2
                  FROM (
                  SELECT
                    *,
                    lr_prob(x1, x2, w0, w1, w2) as predict,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(0)] as delta_w0,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(1)] as delta_w1,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(2)] as delta_w2
                  from data
                  join weights on true
                  )
            )
      )
), 
# --- Step 2
weights_2 as (
      SELECT
        SUM(w0) as w0,
        SUM(w1) as w1,
        SUM(w2) as w2
      FROM (
            SELECT * FROM weights_1
            UNION ALL
            SELECT * FROM (
                  SELECT
                    -0.05 * SUM(delta_w0) / (SELECT COUNT(*) FROM data) as w0,
                    -0.05 * SUM(delta_w1) / (SELECT COUNT(*) FROM data) as w1,
                    -0.05 * SUM(delta_w2) / (SELECT COUNT(*) FROM data) as w2
                  FROM (
                  SELECT
                    *,
                    lr_prob(x1, x2, w0, w1, w2) as predict,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(0)] as delta_w0,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(1)] as delta_w1,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(2)] as delta_w2
                  from data
                  join weights_1 on true
                  )
            )
      )
),
# --- Step 3
weights_3 as (
      SELECT
        SUM(w0) as w0,
        SUM(w1) as w1,
        SUM(w2) as w2
      FROM (
            SELECT * FROM weights_2
            UNION ALL
            SELECT * FROM (
                  SELECT
                    -0.02 * SUM(delta_w0) / (SELECT COUNT(*) FROM data) as w0,
                    -0.02 * SUM(delta_w1) / (SELECT COUNT(*) FROM data) as w1,
                    -0.02 * SUM(delta_w2) / (SELECT COUNT(*) FROM data) as w2
                  FROM (
                  SELECT
                    *,
                    lr_prob(x1, x2, w0, w1, w2) as predict,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(0)] as delta_w0,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(1)] as delta_w1,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(2)] as delta_w2
                  from data
                  join weights_2 on true
                  )
            )
      )
),
# --- Step 4
weights_4 as (
      SELECT
        SUM(w0) as w0,
        SUM(w1) as w1,
        SUM(w2) as w2
      FROM (
            SELECT * FROM weights_3
            UNION ALL
            SELECT * FROM (
                  SELECT
                    -0.02 * SUM(delta_w0) / (SELECT COUNT(*) FROM data) as w0,
                    -0.02 * SUM(delta_w1) / (SELECT COUNT(*) FROM data) as w1,
                    -0.02 * SUM(delta_w2) / (SELECT COUNT(*) FROM data) as w2
                  FROM (
                  SELECT
                    *,
                    lr_prob(x1, x2, w0, w1, w2) as predict,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(0)] as delta_w0,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(1)] as delta_w1,
                    delta_weight(lr_prob(x1, x2, w0, w1, w2), label, x1, x2)[offset(2)] as delta_w2
                  from data
                  join weights_3 on true
                  )
            )
      )
)

SELECT
  0 as step,
  *,
  lr_prob(x1, x2, w0, w1, w2) AS prob,
  lr_prob(x1, x2, w0, w1, w2) > 0.5 AS predict
FROM data
JOIN weights On True

UNION ALL

SELECT
  1 as step,
  *,
  lr_prob(x1, x2, w0, w1, w2) AS prob,
  lr_prob(x1, x2, w0, w1, w2) > 0.5 AS predict
FROM data
JOIN weights_1 On True

UNION ALL

SELECT
  2 as step,
  *,
  lr_prob(x1, x2, w0, w1, w2) AS prob,
  lr_prob(x1, x2, w0, w1, w2) > 0.5 AS predict
FROM data
JOIN weights_2 On True

UNION ALL

SELECT
  3 as step,
  *,
  lr_prob(x1, x2, w0, w1, w2) AS prob,
  lr_prob(x1, x2, w0, w1, w2) > 0.5 AS predict
FROM data
JOIN weights_3 On True

UNION ALL

SELECT
  4 as step,
  *,
  lr_prob(x1, x2, w0, w1, w2) AS prob,
  lr_prob(x1, x2, w0, w1, w2) > 0.5 AS predict
FROM data
JOIN weights_4 On True

ORDER BY STEP ASC, label ASC

↓出力はこうなります（4ステップ目だけを表示）。10個中8個を正しく分類することができました。実はこのデータは線形分離できないデータのためここで最適化はおしまいです。