Databricksにおけるグラフ分析

こちらのサンプルノートブックをウォークスルーします。

https://qiita.com/taka_yayoi/items/a195b1fc636a4613852b

翻訳版のノートブックはこちらです。

https://github.com/taka-yayoi/public_repo_2/blob/master/graph_analysis/graph-analysis-graphframes.scala

# GraphFramesによるグラフ分析

このノートブックでは、[spark-packages.orgで利用できるGraphFramesパッケージ](http://spark-packages.org/package/graphframes/graphframes)を用いた基本的なグラフ分析をウォークスルーします。このノートブックのゴールは、グラフ分析を行うためにどの様にGraphFramesを使うのかを説明することです。[Kaggle](https://www.kaggle.com/benhamner/sf-bay-area-bike-share/downloads/sf-bay-area-bike-share.zip)のベイエリアのバイクシェアのデータを用いてこれを行います。

## グラフ理論とグラフ処理

グラフ処理は多くのユースケースに適用される重要な分析の観点です。基本的に、グラフ理論とグラフ処理は、異なるノードとエッジ間の関係性の定義に関するものです。ノード(vertex)はユニットであり、エッジはそれらの間に定義される関係性です。これは、関係性を理解し、重みづけを行うソーシャルネットワーク分析や、[PageRank](https://en.wikipedia.org/wiki/PageRank)のようなアルゴリズムの実行において非常に役立ちます。

いくつかのビジネスユースケースにおいては、ソーシャルネットワークの中心人物[友達グループで誰が一番人気なのか]、文献ネットワークにおける論文の重要性[どの論文が最も引用されているのか]、Webページのランキングの原因を探すと言ったことが考えられます！

## グラフとバイク移動データ

上で述べた様に、このサンプルではベイエリアのバイクシェアデータを使います。皆様が分析に慣れる様に、すべてのノードをステーションにし、2つのステーションを結ぶエッジが個々の移動になります。これは、*有向*グラフを作成します。

**その他のリファレンス:**
* [Graph Theory on Wikipedia](https://en.wikipedia.org/wiki/Graph_theory)
* [PageRank on Wikipedia](https://en.wikipedia.org/wiki/PageRank)

## データの準備

1. KaggleからBay Area Bike Share[データ](https://www.kaggle.com/benhamner/sf-bay-area-bike-share)をダウンロードして解凍します。サードパーティの認証を用いてKaggleにサインインするか、Kaggleアカウントを作成してサインインする必要があります。
1. [Explore and create tables in DBFS](https://docs.databricks.com/data/data-tab.html)で説明されている方法で`station.csv`と`trip.csv`をアップロードします。テーブル名は`station_csv`と`trip_csv`になります。

![Screen Shot 2022-11-12 at 12.19.12.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/bb327bf2-ae37-c47c-76ba-f3b50d486895.png)

# データフレームの作成

上の[データの準備](#データの準備)で作成したテーブルを指定してデータを読み込みます。

```scala:Scala
val bikeStations = spark.sql("SELECT * FROM takaakiyayoi_db.station_csv")
val tripData = spark.sql("SELECT * FROM takaakiyayoi_db.trip_csv")
```

```scala:Scala
display(bikeStations)
```
![Screen Shot 2022-11-12 at 15.10.35.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/13703d10-5bca-9ee2-afc1-716287c248cc.png)

```scala:Scala
display(tripData)
```
![Screen Shot 2022-11-12 at 15.11.09.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b374c5b1-044c-de1e-a1f2-3b1568a85140.png)

適切な型が適切なカラムに割り当てられていることを確認するために、正確なスキーマを確認することは多くの場合有用です。

```scala:Scala
bikeStations.printSchema()
tripData.printSchema()
```
![Screen Shot 2022-11-12 at 15.11.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ad6e6306-3ff6-39bb-2909-5debcf57b916.png)

# インポート

続ける前にいくつかのインポートが必要です。データフレームの操作を簡単にするさまざまなSQL関数ををインポートし、GraphFramesに必要な全てをインポートします。

```scala:Scala
import org.apache.spark.sql._
import org.apache.spark.sql.functions._

import org.graphframes._
```

# グラフの構築

データをインポートした後に必要なのはグラフの構築です。これを行うために2つのことが必要です。ノード(vertex)の構造を構築し、エッジの構造を構築します。GraphFramesの素晴らしいところは、このプロセスが信じられないほどシンプルだということです。必要なのは、ノードのテーブルの別個の**id**値を取得し、エッジテーブルの起点と終点のステーションをそれぞれ**src**と**dst**に変更するということです。これは、GraphFramesにおけるノードとエッジに必要な決まり事です。

```scala:Scala
val stationVertices = bikeStations
  .distinct()

val tripEdges = tripData
  .withColumnRenamed("start_station_name", "src")
  .withColumnRenamed("end_station_name", "dst")
```

```scala:Scala
display(stationVertices)
```
![Screen Shot 2022-11-12 at 15.14.16.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/99766637-97c0-20b2-944a-9f8d1dd89e88.png)

```scala:Scala
display(tripEdges)
```
![Screen Shot 2022-11-12 at 15.14.50.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cd9473ee-1cf5-5def-f66b-cd6440e97aa8.png)

これでグラフを構築できます。

また、ここでグラフへの入力データフレームをキャッシュしておきます。

```scala:Scala
val stationGraph = GraphFrame(stationVertices, tripEdges)

tripEdges.cache()
stationVertices.cache()
```

```scala:Scala
println("Total Number of Stations: " + stationGraph.vertices.count)
println("Total Number of Trips in Graph: " + stationGraph.edges.count)
println("Total Number of Trips in Original Data: " + tripData.count)// sanity check
```
![Screen Shot 2022-11-12 at 15.15.46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c0c9f19e-3dff-5506-8a2a-49d7019b7525.png)

# ステーション間の移動

よく尋ねられる質問は、データセットにおいて最も共通している目的地が何かということです。グルーピングオペレーターとエッジのカウントを組み合わせることで、これを行うことができます。これは、エッジを除外した新たなグラフを作り出し、意味的に同じエッジすべての合計値となります。この様に考えてみましょう: 全く同じステーションAからステーションBへの移動回数が存在し、単にこれらをカウントするだけです！

以下のクエリーでは、最も共通するステーション間移動を抽出し、トップ10を表示しています。

```scala:Scala
val topTrips = stationGraph
  .edges
  .groupBy("src", "dst")
  .count()
  .orderBy(desc("count"))
  .limit(10)

display(topTrips)
```
![Screen Shot 2022-11-12 at 15.17.01.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/af89690c-bc36-8ab4-9fdb-2d3e3bb72ac6.png)

上の結果から、特定のノードがカルトレインのステーションであることが重要であることがわかります！これらは自然な接続点であり、車を使わない方法で、これらのバイクシェアプログラムを用いてAからBに移動するには最も人気のある利用法なのでしょう！

# 入次数(in degree)と出次数(out degree)

この例では有向グラフを使っていることを思い出してください。これは、移動には方向があること - ある地点からある地点へ - を意味します。これによって、あなたが活用できる分析はさらにリッチなものになります。特定のステーションに移動する数や、特定のステーションから移動する数を見つけ出すことができます。

通常、この情報で並び替えを行い、インバウンドとアウトバウンドの移動が多いステーションを見つけ出すことができます！詳細に関しては、[Vertex Degrees](http://mathworld.wolfram.com/VertexDegree.html)の定義をチェックしてみてください。

このプロセスを定義したので、次に進んでインバウンドとアウトバウンドの移動が多いステーションを見つけましょう。

```scala:Scala
val inDeg = stationGraph.inDegrees
display(inDeg.orderBy(desc("inDegree")).limit(5))
```
![Screen Shot 2022-11-12 at 15.17.53.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/007f3d1c-e127-d99e-2f5d-7936fce9419c.png)

```scala:Scala
val outDeg = stationGraph.outDegrees
display(outDeg.orderBy(desc("outDegree")).limit(5))
```
![Screen Shot 2022-11-12 at 15.18.32.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4af73c24-c0dd-c723-0bcd-e20bc262796e.png)

もう一つの興味深い質問は、入次数が最も高いが、出次数が少ないステーションがどれかということです。すなわち、どのステーションが純粋な移動のシンクになっているかということです。移動がそこで終了しますが、ほとんどそこから移動を開始しない場所です。

```scala:Scala
val degreeRatio = inDeg.join(outDeg, inDeg.col("id") === outDeg.col("id"))
  .drop(outDeg.col("id"))
  .selectExpr("id", "double(inDegree)/double(outDegree) as degreeRatio")

degreeRatio.cache()
  
display(degreeRatio.orderBy(desc("degreeRatio")).limit(10))
```
![Screen Shot 2022-11-12 at 15.19.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/174bca8e-bfc7-e5b9-4daf-df37120fdc3b.png)

出次数に対する入次数の比率が最も低いステーションを得ることで、同様のことを行うことができます。これは、そのステーションからの移動開始が多いが、移動がそこで終わることがあまりないことを意味します。これは、基本的に上で見たのと逆の内容になります。

```scala:Scala
display(degreeRatio.orderBy(asc("degreeRatio")).limit(10))
```
![Screen Shot 2022-11-12 at 15.20.04.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/ae91c8a8-bbb2-39a1-5187-e313dd10e1c3.png)

上の分析から得られる結論は比較的わかりやすいものです。高い値は出てくるよりも入る移動が多いことを意味し、低い値はそのステーションからの移動開始が多いが、移動終了は少ないということです！

このノートブックから何かしらの価値を得ていただけたら幸いです！グラフ構造は探し始めるとどこにでもあり、GraphFramesが分析を容易にしてくれることを願っています！

### Databricks 無料トライアル

[Databricks 無料トライアル](https://databricks.com/jp/try-databricks)