Databricksにおけるグラフ解析のPythonサンプルのウォークスルー

Sparkでは、グラフ構造を容易に取り扱うことができる[GraphFrames](https://github.com/graphframes/graphframes)という機能を利用できます。[こちら](https://qiita.com/taka_yayoi/items/a436ffe306b6d23d432e)のソリューションアクセラレータでもBoM(部品表)をデータ分析で活用しています。

2年前にこちらの記事を書いた際には、Scalaのサンプルしかなかったのですが、Pythonのサンプルができていました。嬉しいので、翻訳しつつウォークスルーします。

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

https://qiita.com/taka_yayoi/items/6d90f8cb8bd661e8e26b

サンプルノートブックはこちらにあります。

https://docs.databricks.com/ja/integrations/graphframes/index.html#graphframes-python-notebook

翻訳版はこちら。

https://github.com/taka-yayoi/public_repo_2/blob/master/graphframes_python/graphframes-user-guide-py.py

# GraphFramesユーザーガイド (Python)

このノートブックでは、[GraphFrames User Guide](https://graphframes.github.io/graphframes/docs/_site/user-guide.html)の例をデモンストレーションします。

## 要件

このノートブックでは、機械学習ランタイムが必要となります。

```py
from functools import reduce
from pyspark.sql import functions as F
from graphframes import GraphFrame
```

グラフを可視化した方がわかりやすいので、以下を参考にさせていただきました。

[python \- PYSPARK: how to visualize a GraphFrame? \- Stack Overflow](https://stackoverflow.com/questions/45720931/pyspark-how-to-visualize-a-graphframe)

```py
import networkx as nx
import matplotlib.pyplot as plt

def PlotGraph(edge_list):
    Gplot = nx.Graph()
    for row in edge_list.select("src", "dst").take(1000):
        Gplot.add_edge(row["src"], row["dst"])

    plt.subplot(121)
    nx.draw(Gplot, with_labels=True, font_weight="bold")
```

## GraphFramesの作成

ユーザーは頂点(vertex)とエッジ(edge)のデータフレームからGraphFramesを作成できます。

* Vertexデータフレーム: Vertexデータフレームには、グラフにおけるそれぞれの頂点に対するユニークなIDを示す"id"という名前の特殊なカラムを含める必要があります。
* Edgeデータフレーム: Edgeデータフレームには、2つの特殊なカラムが必要です: "src" (エッジのソースとなる頂点のID) と "dst" (エッジのディスティネーションとなる頂点のID)です。

両方のデータフレームには、任意のその他のカラムを含めることができます。これらのカラムでは、頂点やエッジの属性を表現することができます。

はじめに頂点を作成します:

```py
vertices = spark.createDataFrame([
    ("a", "Alice", 34),
    ("b", "Bob", 36),
    ("c", "Charlie", 30),
    ("d", "David", 29),
    ("e", "Esther", 32),
    ("f", "Fanny", 36),
    ("g", "Gabby", 60)],
    ["id", "name", "age"])
```

次に幾つかのエッジを作成します:

```py
edges = spark.createDataFrame([
    ("a", "b", "friend"),
    ("b", "c", "follow"),
    ("c", "b", "follow"),
    ("f", "c", "follow"),
    ("e", "f", "follow"),
    ("e", "d", "friend"),
    ("d", "a", "friend"),
    ("a", "e", "friend")], 
    ["src", "dst", "relationship"])
```

これらの頂点とエッジからグラフを作成します:

```py
g = GraphFrame(vertices, edges)
print(g)
```
```
GraphFrame(v:[id: string, name: string ... 1 more field], e:[src: string, dst: string ... 1 more field])
```

可視化します。以降はこちらを見ながらコードを実行するとわかりやすいかと思います。

```py
PlotGraph(g.edges)
```
![download.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/09f226e2-bb40-dfbf-1e3f-1442b542fd00.png)

```py
# このサンプルnのグラフはGraphFramesパッケージでも提供されています。
from graphframes.examples import Graphs
same_g = Graphs(spark).friends()
print(same_g)
```

## 基本的なグラフとデータフレームのクエリー

GraphFramesでは、ノードの度数のようないくつかのシンプルなグラフクエリーを提供します。

また、GraphFramesはグラフを頂点とエッジデータフレームのペアとして表現するので、頂点とエッジのデータフレームに対して直接パワフルなクエリーを容易に実行することができます。これらのデータフレームは、GraphFrameの`vertices`や`edges`フィールドから利用できます。

```py
display(g.vertices)
```
![Screenshot 2024-08-06 at 17.19.33.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/78910c2c-8a02-5406-14cb-dd5a53554d2d.png)

```py
display(g.edges)
```
![Screenshot 2024-08-06 at 17.19.49.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/05e5ff55-9ff3-755a-a5b2-0908ab3590df.png)

頂点の内向きの度数:

```py
display(g.inDegrees)
```
![Screenshot 2024-08-06 at 17.20.15.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cd3dc8eb-08d7-4c7e-6f1e-97b9b11dc1a5.png)

頂点の外向きの度数:

```py
display(g.outDegrees)
```
![Screenshot 2024-08-06 at 17.20.43.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b8e38876-e7fe-e713-05b4-d2dae05f9c3b.png)

頂点の度数:

```py
display(g.degrees)
```
![Screenshot 2024-08-06 at 17.23.08.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/970ca477-c056-a7b3-c8bf-4b5f79c960c0.png)

verticesデータフレームに直接クエリーを実行できます。例えば、グラフにおいて最も若い人物の年齢を特定できます:

```py
youngest = g.vertices.groupBy().min("age")
display(youngest)
```
![Screenshot 2024-08-06 at 17.23.46.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/9c18de07-1c88-11d2-9ab3-d155c32dac85.png)

また、edgesデータフレームにクエリーを行うことができます。例えば、グラフにおける *follow* リレーションシップの数をカウントします:

```py
numFollows = g.edges.filter("relationship = 'follow'").count()
print("The number of follow edges is", numFollows)
```
```
The number of follow edges is 4
```

## モチーフの特定

モチーフを用いることで、エッジと頂点を含むより複雑なリレーションシップを構築できます。以下のセルでは、両方向で接続されている頂点とエッジのペアを特定しています。結果は、モチーフのキーによって指定されるカラム名を持つデータフレームとなります。

APIの詳細に関しては、[GraphFrame User Guide](https://graphframes.github.io/graphframes/docs/_site/user-guide.html#motif-finding)をご覧ください。

```py
# 両方向で接続されている頂点とエッジのペアを検索します。
motifs = g.find("(a)-[e]->(b); (b)-[e2]->(a)")
display(motifs)
```
![Screenshot 2024-08-06 at 17.24.51.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/82c09c32-18ed-bf77-fb4d-46f3dc521f68.png)

結果はデータフレームとなるので、モチーフをベースにしてより複雑なクエリーを構築することができます。以下のセルでは、どちらかが30歳を上回っているすべての相互リレーションシップを検索します:

```py
filtered = motifs.filter("b.age > 30 or a.age > 30")
display(filtered)
```
![Screenshot 2024-08-06 at 17.25.16.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/75fae7f9-a48a-3dc0-cbd0-ed9518b1c8d3.png)

### ステートフルなクエリー

上の例のように、ほとんどのモチーフクエリーはステートレスで表現が容易です。次の例では、モチーフのパスを通じて状態を運ぶより複雑なクエリーをデモンストレーションします。このようなクエリーは、GraphFrameモチーフの検索と、検索結果のデータフレームのカラムに対して適用される後続のオペレーションのフィルターを組み合わせることで表現できます。

例えば、一連の関数によって定義されるいくつかのプロパティを持つ、4つの頂点のチェーンを特定したいものとします。すなわち、4つの頂点のチェーン`a->b->c->d`において、このような複雑なフィルターにマッチするチェーンのサブセットを特定します:

* パスの状態の初期化
* 頂点aに基づいて状態を更新
* 頂点bに基づいて状態を更新
* cやdも同様

最終的な状態が何かしらの条件に合致したら、フィルターはチェーンを受け入れます。

以下のコードスニペットではこのプロセスを説明しています。このコードでは、3つのエッジのうち少なくとも2つが"friend"リレーションシップである、4つの頂点のチェーンを特定します。この例では、状態は現時点での"friend"エッジのカウントとなります。通常、これはデータフレームのカラムとなります。

```py
# 4つの頂点を持つチェーンを特定します。
chain4 = g.find("(a)-[ab]->(b); (b)-[bc]->(c); (c)-[cd]->(d)")

# 状態(cnt)とともにシーケンスをクエリー
#  (a) モチーフの次の要素に基づいて状態を更新するメソッドを定義します。
def cumFriends(cnt, edge):
    relationship = F.col(edge)["relationship"]
    return F.when(relationship == "friend", cnt + 1).otherwise(cnt)

#  (b) モチーフの要素のシーケンスに対してメソッドを適用するためにシーケンスオペレーションを活用します。
#   この場合、要素は3つのエッジとなります。
edges = ["ab", "bc", "cd"]
numFriends = reduce(cumFriends, edges, F.lit(0))
    
chainWith2Friends2 = chain4.withColumn("num_friends", numFriends).where(numFriends >= 2)
display(chainWith2Friends2)
```
![Screenshot 2024-08-06 at 17.26.27.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/4500ab7a-5b69-b32a-a3a3-44b1b7a5395e.png)

## サブグラフ

GraphFramesはエッジや頂点に基づいたフィルタリングによってサブグラフを構築するAPIを提供します。これらのフィルターは、以下のような30歳を超える人、かつ、友達も30歳を超えている人のみを含むサブグラフのように、組み合わせることが可能です。

```py
g2 = g.filterEdges("relationship = 'friend'").filterVertices("age > 30").dropIsolatedVertices()
```
```py
display(g2.vertices)
```
![Screenshot 2024-08-06 at 17.27.22.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/5bbf4ae9-28ee-e8c6-b12f-ef57b50c9cb9.png)

```py
display(g2.edges)
```
![Screenshot 2024-08-06 at 17.27.48.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/187a4994-1ec7-66ed-6198-118074a95eca.png)

## 標準的なグラフアルゴリズム

GraphFramesでは、数多くの標準的なグラフアルゴリズムがビルトインされています:

* Breadth-first search (BFS)
* Connected components
* Strongly connected components
* Label Propagation Algorithm (LPA)
* PageRank (regular and personalized)
* Shortest paths
* Triangle count

### Breadth-first search (BFS)

age < 32のユーザーから"Esther"を検索します。

```py
paths = g.bfs("name = 'Esther'", "age < 32")
display(paths)
```
![Screenshot 2024-08-06 at 17.28.37.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/3dff6766-4bb4-b730-9ca2-25ec717ab878.png)

エッジのフィルターや最大パス長で検索を制限することもできます。

```py
filteredPaths = g.bfs(
    fromExpr = "name = 'Esther'",
    toExpr = "age < 32",
    edgeFilter = "relationship != 'friend'",
    maxPathLength = 3)
display(filteredPaths)
```
![Screenshot 2024-08-06 at 17.29.25.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/2e7aa1b0-6ad0-53c1-fc25-79da1dadef2e.png)

### 接続されたコンポーネント(Connected components)

それぞれの頂点で接続されたコンポーネントのメンバーシップを計算し、コンポーネントIDが割り当てられたそれぞれの頂点を持つデータフレームを返却します。GraphFramesのConnected componentsの実装では、パフォーマンスを改善するためにチェックポイントを活用することができます。

```py
sc.setCheckpointDir("/tmp/graphframes-example-connected-components")
result = g.connectedComponents()
display(result)
```
![Screenshot 2024-08-06 at 17.29.54.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/b9a83443-9359-7bb8-4f63-652af2a9206b.png)

### 強く接続されたコンポーネント(Strongly connected component)

それぞれの頂点のstrongly connected component (SCC)を計算し、当該の頂点を含むSCCに割り当てられたそれぞれの頂点を持つデータフレームを返却します。

```py
result = g.stronglyConnectedComponents(maxIter=10)
display(result.select("id", "component"))
```
![Screenshot 2024-08-06 at 17.30.24.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/8810c260-5246-1101-d75b-14471e60c06b.png)

### ラベルの伝播

ネットワークにおけるコミュニティを検知するために、静的なラベル伝播アルゴリズムを実行します。

ネットワークにおけるそれぞれのノードは、最初に自身のコミュニティに割り当てられます。すべてのスーパーステップにおいて、ノードは全ての隣人にコミュニティへの協力関係を依頼し、到着するメッセージから最も高頻度なコミュニティの協力依頼に自身の状態を更新します。

LPAは、グラフにおける標準的なコミュニティ検知アルゴリズムです。これは、以下の点に考慮する必要がありますが、計算量的には安価なものとなっています。

1. 収束が保証されません 
1. つまらない回答になる場合があります(全てのノードが単独なコミュニティとして識別される) 

```py
result = g.labelPropagation(maxIter=5)
display(result)
```
![Screenshot 2024-08-06 at 17.31.26.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/7acb997c-5410-d266-be46-2dda972f9e3a.png)

### PageRank

接続性に基づいてグラフにおいて重要な頂点を特定します。

```py
results = g.pageRank(resetProbability=0.15, tol=0.01)
display(results.vertices)
```
![Screenshot 2024-08-06 at 17.31.52.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d97c2a11-2129-d096-8286-874cd9490c35.png)

```py
display(results.edges)
```
![Screenshot 2024-08-06 at 17.32.20.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/cbc55120-5364-ca78-0b68-9dc3d6493cca.png)

```py
# 固定回数のイテレーションでPageRankを実行します。
results = g.pageRank(resetProbability=0.15, maxIter=10)
display(results.vertices)
```
![Screenshot 2024-08-06 at 17.32.39.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/c52d5900-bc23-ed32-c321-04a3a0c204fb.png)

```py
# 頂点"a"にパーソナライズしたPageRankの実行
results = g.pageRank(resetProbability=0.15, maxIter=10, sourceId="a")
display(results.vertices)
```
![Screenshot 2024-08-06 at 17.33.00.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/a8742689-7d27-517e-f670-a22ac8922dfe.png)

### 最短パス

頂点IDで指定されるランドマークの頂点のセットへの最短パスを計算します。

```py
results = g.shortestPaths(landmarks=["a", "d"])
display(results)
```
![Screenshot 2024-08-06 at 17.33.31.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/d9d33065-f734-1f98-7f42-91682996fad4.png)

### 三角形のカウント

それぞれの頂点を通過する三角形の数を計算します。

```py
results = g.triangleCount()
display(results)
```
![Screenshot 2024-08-06 at 17.34.14.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/1168882/6af01cfc-0734-9e21-88a2-c5d3cc835972.png)

### はじめてのDatabricks

[はじめてのDatabricks](https://qiita.com/taka_yayoi/items/8dc72d083edb879a5e5d)

### Databricks無料トライアル

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