# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# # Spark GraphFrames

import numpy as np

from pyspark.sql import SparkSession
spark = SparkSession.builder.getOrCreate()

from pyspark.sql.types import *
from graphframes import *

import graphframes

# **Notes on installing GraphFrames**
#
# I needed to put the GraphFrames jar file into the Spark jars director for it to work.
#
# ```bash
# # cd /usr/local/spark/jars/
# wget http://dl.bintray.com/spark-packages/maven/graphframes/graphframes/0.7.0-spark2.4-s_2.11/graphframes-0.7.0-spark2.4-s_2.11.jar
# ```

import networkx as nx
from networkx.drawing.nx_pydot import graphviz_layout

# %matplotlib inline
import matplotlib.pyplot as plt

G = nx.les_miserables_graph()

plt.figure(figsize=(14, 6))
nx.draw_kamada_kawai(
    G, 
    with_labels=True, 
    node_size=500, 
    node_color='lightblue' 
)

np.random.seed(123)

edges = [(a, b, np.random.randint(1, 4)) for (a, b) in nx.edges(G)]

nodes = [(name, name, np.random.randint(15, 60)) for name in nx.nodes(G)]

edges[:3]

nodes[:3]

# ## Creation of a GraphFrame
#
# - Need to provide a DataFrame of vertices, and a DataFrame of edges

# ### Vertices must contain an id column

v = spark.createDataFrame(nodes, ['id', 'name', 'age'])

# ### Edges must have src and dst columns that contain vertex ids

e = spark.createDataFrame(edges, ['src', 'dst', 'wt'])

g = GraphFrame(v, e)

# ## Basic information

g

g.vertices.show(5)

g.edges.show(5)

# ## Search

from pyspark.sql.functions import col, desc

g.vertices.show(5)

# ### Can use regular DataFrame functions

g.vertices.filter(col('name').startswith('C')).show(5)

g.edges.groupBy('wt').count().show()

# ## Motifs

g.find('(a)').show(5)

g.find('(a)-[e]->(b)').show(5, truncate=False)

g.find('(a)-[e1]->(b); (b)-[e2]->(c)').select('e1', 'e2').show(5, truncate=False)

g.find('(a)-[]->(b); !(b)-[]->(a)').show(5, truncate=False)

g.find('(a)-[]->(b); (b)-[]->(a)').show(5, truncate=False)

# ## Extracting subgraphs

g1 = g.filterVertices("age > 30").filterEdges('wt < 3').dropIsolatedVertices()

g.vertices.count()

g1.vertices.count()

# ## Pathfinding

path = g.bfs("name = 'Cosette'", "age > 30")

path.show(truncate=False)

path = g.bfs("name = 'Cosette'", "age = 50")

path.show(truncate=False)

paths = g.shortestPaths(landmarks = ['Cosette'])

paths.sort('id').show(100)

# ## Centrality

g.degrees.show(5)

g.inDegrees.show(5)

g.outDegrees.show(5)

res = g.pageRank(resetProbability=0.15, maxIter=5)

res.vertices.sort(desc('pagerank')).show(5, truncate=False)

res = g.pageRank(resetProbability=0.15, maxIter=5, sourceId='Cosette')

res.vertices.sort(desc('pagerank')).show(5, truncate=False)

# ## Community detection

sc = spark.sparkContext
sc.setCheckpointDir("checkpoint")

res = g1.connectedComponents()

res.select('component').distinct().show()

res.orderBy('component').show(5, truncate=False)

res = g1.stronglyConnectedComponents(maxIter=5)

res.groupBy('component').count().filter('count > 1').show()

res.orderBy('component').show(10, truncate=False)

res = g.triangleCount()

res.show(5, truncate=False)

res = g.labelPropagation(maxIter=5)

res.groupBy('label').count().filter('count > 1').show()





StarcoderdataJupyterScriptsDedupFiltered