Azureデータ分析入門 #4 【ツール比較  Databricks編】

## はじめに
ツール比較の3つ目として、**Databricks** を用いて簡易集計及び可視化を行い、発生した費用を概算します。


#### 連載目次
[Azureデータ分析入門 #1 【はじめに】](https://qiita.com/Catetin0310/items/a7ab83069fec5352b535)
[Azureデータ分析入門 #2 【ツール比較 Excel編】](https://qiita.com/Catetin0310/items/d32c53305ab579d47950)
[Azureデータ分析入門 #3 【ツール比較 Azure Notebook編】](https://qiita.com/Catetin0310/items/ae3c5f3dca5b3315896d)
[Azureデータ分析入門 #4 【ツール比較 Databricks編】](https://qiita.com/Catetin0310/items/b6030d3ad0fde16e57ab) → 本記事はこちら
[Azureデータ分析入門 #5 【Databricks → Power BI Desktop】](https://qiita.com/Catetin0310/items/eb8431f38a51641b0db1)
[Azureデータ分析入門 #6 【CSVデータ → Power BI サービス】](https://qiita.com/Catetin0310/items/4af00df1f0af5feda1bb)
Azureデータ分析入門 #7 【AutoML でタイタニック号の生存者予測】


## Databricksとは？
### 概要
[以前の記事](https://qiita.com/Catetin0310/items/d515fc3fdff68b54f408)でも取り上げましたが、こんな特徴のデータ分析基盤プラットフォームです。

- **Apache Spark環境を数分でセットアップ**
    - 分析環境がすぐに作れる
    - スケーラブル
    - 自動でスケーリング可能
    - Sparkを作った人達が、Spark用にカリカリにチューニングしたプラットフォーム

- **Notebookをバッチスクリプトとして使える**
    - 例えば、
    - データストリームに対して、Aの処理を一週間に一回かけて、Bのテーブルに追記する、
    - という処理が必要な場合、Aの処理を記載したNotebookを、バッチジョブとして設定できる

- **バージョン管理が楽**
    - いわゆるタイムマシーン機能が使える
    - 複数人での作業がしやすい
  
- **マジックコマンドが便利**
    - デフォルト言語としてPythonを設定しても、セルの頭にマジックコマンド(```%sql```)でクエリが書ける

- **簡単に可視化**
    - 発行したクエリをそのままグラフ化
    - 別途ライブラリのインポート不要

- **オートターミネーションで安心**
    - クラスタをxx分未使用の場合には停止(=課金されない) といった設定ができる
    - 高額なクラスタを立ち上げっぱなしで、気が付いたらものすごい課金が…みたいなケースを回避できる
    - 今回発生した費用のまとめは後述します

### 設定
[**こちら**](https://docs.microsoft.com/ja-jp/azure/azure-databricks/quickstart-create-databricks-workspace-portal)のドキュメントを参照いただき、

- Azure Portal へのサインイン
- Spark クラスタの作成
- notebookの作成

まで行ったら準備完了です。

### 操作のきほん
Notebook自体の操作は、Jupyter Notebookとほとんど同じです。結構直感的に操作できますが、迷ったときは[公式ドキュメント](https://docs.microsoft.com/ja-jp/azure/azure-databricks/)を参照しましょう。

## 手順
Azure Notebookの回でやったように、Sparkで中間テーブルを作成します。そこから先は、SQLオンリーでグラフ化まで行います。

#### 使用するデータのインポート
ストリーミングデータや大規模なテーブルを接続するのであれば、データの保存先は、**Blob Storage Gen 2** や **Delta Lake** が候補になります。今回は、そこまで大きくないCSVデータx3ですので、Databricksにテーブルを作成します。

ホーム画面の下記の場所に、CSVデータをドラッグアンドドロップ
![00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/363781/8334a55e-46b2-a6a9-25f5-6511719d3694.png)

Create Table with UIをクリック
![01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/363781/9cedece9-ce44-df65-1254-87904c0f9c57.png)

作成したクラスタを選択し、Preview Tableをクリック
![02.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/363781/054fc405-1f07-4954-465b-a96b3c7a7519.png)

デフォルトだと、列名やスキーマはこのようになっています
![03.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/363781/bf4a1bad-3aa7-e5b3-afdc-c903567ad71a.png)

以下のオプションを指定し、Create Tableをクリック。

First Row is Header オプション: 一行目を列名として読み込み
Infer Schema オプション: スキーマを類推

![04.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/363781/8d317183-310c-b945-193f-2e7392160e3e.png)

しばらくするとテーブルの作成が完了します。CSVファイル3つそれぞれでこれらの作業を行います。Data > Databases > Tables でアップロードしたデータを確認できます。

(これらのテーブルは内部的には Spark Dataframe として格納されています。クエリを書くためには後述の View を作ってあげる必要があります)

![2019-09-13_16h30_36.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/363781/6f0f3d2d-5f7f-ad5e-7eab-6fbd3aec83e8.png)


#### ライブラリのインポート
今回の用途であればこれだけでOKです。

```python
from pyspark.sql.functions import col, desc, to_timestamp, month, year, dayofweek, hour
```


#### データフレームの読み込み
先ほど作成したデータフレームの名称を変数に格納。

```python
orders = 'olist_order_items_dataset_csv'
items = 'olist_products_dataset_csv'
translation = 'product_category_name_translation_csv'
```

それぞれのデータフレームをSparkで読み込みます。

```python
t1 = spark.read.table(orders)
t2 = spark.read.table(items)
t3 = spark.read.table(translation)
```

日付データの書式をタイムスタンプに、結合に使用するキーの名称を整理しておきます。

```python
t1 = t1.withColumn('time', to_timestamp(t1.shipping_limit_date, 'yyyy/MM/dd HH:mm'))
t2 = t2.withColumnRenamed('product_id', 'product_id2')
t3 = t3.withColumnRenamed('product_category_name', 'product_category_name2')
```


#### データフレームの結合
フレーム結合のために、必要な列名を確認。

```python
t1.printSchema()
t2.printSchema()
t3.printSchema()
```

とりあえず3つのデータフレームを結合します。カテゴリの英語名が長いので、名称を変えておきます。

```python
temp_t = t1\
.join(t2, t1.product_id == t2.product_id2, 'inner')\
.join(t3, t2.product_category_name == t3.product_category_name2, 'inner')\
.withColumnRenamed('product_category_name_english', 'category')\
```

結合後のデータフレームのスキーマを確認しておきます。

```python
temp_t.printSchema()
```

#### ビューの作成
集計用の一時ビューとして、```temp_view``` を作成します。

```python
temp_t.select('order_id'\
              , 'product_id'\
              , 'category'\
              , 'price'\
              , 'freight_value'\
              , 'time'\
              , hour('time').alias('hour')\
              , dayofweek('time').alias('dayofweek')\
              , month('time').alias('month')\
              , year('time').alias('year'))\
.createOrReplaceTempView('temp_view')
```


#### SQL で集計 → 可視化
以上の操作で、マジックコマンド ```%sql``` を使用してクエリを直接書けるようになりました。

```python
%sql
SELECT * FROM temp_view
```

テーブルの詳細は以下のコードで確認できます。

```python
%sql
DESCRIBE temp_view
```



マジックコマンドを使用してのクエリでも、一時ビューを作成することができます。集計結果を格納するビュー ```agged_view``` をこの方法で作成してみましょう。(今回は使用しませんが、カテゴリ別売上のランキングのカラムを追加しています)

```python
%sql
CREATE OR REPLACE TEMPORARY VIEW agged_view AS
  SELECT
      category as cat
      , sum(price) / count(order_id) as cat_price_ave
      , sum(freight_value) / count(order_id) as cat_freight_ave
      , sum(freight_value) / (sum(freight_value) + sum(price)) as category_freight_ratio
      , RANK () OVER (ORDER BY sum(price) DESC) as ranking
  FROM
      temp_view
  GROUP BY
      category
```

agged_view の一部を抜き出し、可視化します。平均単価と送料の割合が同じくらいのスケールに収まるように、送料の割合を1000倍しておきます。

```python 
%sql
SELECT
  cat
  , category_freight_ratio * 1000 as freight_ratio_x_1000
  , cat_price_ave
FROM agged_view
WHERE ranking > 0 AND ranking <= 10
ORDER BY cat_price_ave DESC
```

結果はこちら。デフォルトだと表が出力されます。下のボタンでクエリの結果をそのまま可視化可能。
![2019-09-24_14h18_44.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/363781/1f0317a5-6ccf-e336-f528-51e1d5f9ec7f.png)



こんな感じです。アドホックな分析にとても便利！
![zeppelin.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/363781/9aaadac4-c065-90b6-2c6e-a1d4e6bd0b79.gif)



## 費用
### 料金体系
クラスターの起動時間に応じて、分単位で、以下のマシンリソースに対して課金されます。 (詳細は[**こちら**](https://azure.microsoft.com/ja-jp/pricing/details/databricks/)) 

|名称|概要|
|---|---|
|Virtual Machine|通称 VM。Azure上の仮想マシン|
|Databricks Unit|通称 DBU。VMインスタンスに基づくDatabricks計算資源の単位|
|ストレージ|分析対象データを保管するストレージ|

料金表だけを見ていると意識に上がりずらいのですが、**Databricks Unit** には **Driver** と**Worker**の2タイプがあり、クラスタを立ち上げると両方が起動され、課金されます。それぞれにインスタンスを割り当てて処理をしてもらうわけですが、当然、性能の高いインスタンス(= 高いDBU値) ほど、高コストになります。
今回使用したのはもっとも安価な 0.5 DBU のインスタンス。

|名称|概要|
|---|---|
|Driver|1台必要。分散処理をマネージメントするDBU|
|Worker|1台以上必要。実際に処理を行うDBU。最小数と最大数を設定でき、オートスケール可能|

### 今回の費用概算
今回の簡易分析・可視化で、どのくらいの費用が発生したか概算してみましょう。以下は今回使用したクラスタの設定画面。費用に大きく響いてくるところなので、本格的に分析基盤を立ち上げる際には、分析対象と内容に応じて、必要十分な計算資源を割り当てるように留意する必要があります。

![2019-09-18_13h30_00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/363781/09e256d8-22d0-f849-2dd6-34ed2ad6a6cb.png)

設定は以下の通り。
クラスタのJob履歴を見てみたところ、Workerのスケーリングはしなかったようなので、クラスタの起動中は、0.5 DBU + 0.5 DBU で合計 1 DBU 使ってる計算になります。

- リージョン： 米国西部 2
- ワークロード：データ分析
- レベル：Standard
- Driverタイプ：F4s (0.5 DBU)
- Workerタイプ：F4s (0.5 DBU)
    - 最小ワーカー数: 1
    - 最大ワーカー数: 4
- 自動停止
    - 15分操作がなかったらクラスタを停止

分単位の課金なので、Cluster の Event Log から、ここ2日間で何分間クラスタを起動したかざっくり見てみます。3回起動/停止しているようです。合計約70分。

![2019-09-18_14h36_17.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/363781/0e0131af-4fd4-09af-9af0-cf81ea339243.png)

Databricks の料金表を参照し、70分間インスタンスを起動した今回の費用概算と、仮に48時間インスタンスを立てたままにした場合の料金を比較してみました。

当たり前ですが、オートターミネーションした場合と継続利用では、かなりの金額差がありますね。

![costlast.jpg](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/363781/4d0a0382-c106-8ba8-2de5-5bf042e9a591.jpeg)


分析プラットフォームサービスは、どうしてもマシンスペックでの比較がされやすい傾向にあります。

しかし、実運用に則して考えると、Databricks の費用対効果の高さはピカイチじゃないかと思います。


## まとめ
最後にDatabricksの長所をおさらい。

- すぐApache Spark環境を作れる
- Notebookをバッチスクリプトとして使える
- バージョン管理が楽
- マジックコマンドが便利
- 簡単に可視化
- オートターミネーション

アドホックな分析をするなら、SQLをそのままかけて、すぐに可視化できて、クラスタのオートターミネーションが利用できる Databricks が最高に便利。

Databricks、簡易的な可視化ならできるけど、ちょっと凝ったグラフは？BIとの接続は？ということで、次回は **Databricks** を **Power BI** につなげてみようと思います。お楽しみに！

## 参考サイト
[クイック スタート: Azure portal を使用して Azure Databricks 上で Spark ジョブを実行する](https://docs.microsoft.com/ja-jp/azure/azure-databricks/quickstart-create-databricks-workspace-portal) 
[Databricksで分析業務がはかどっている話](
https://qiita.com/paulxll/items/1962e806cdc8a96068c1)
[平成最後の1月ですし、Databricksでもやってみましょうか](https://www.slideshare.net/tottokug/1databricks)