Graphframe

Prerequisites

Technical Requirements

• Apache Spark 2.3+ installed
• Python 3.6+ installed
• PySpark installed
• GraphFrames package

Knowledge Requirements

• Basic understanding of graph theory concepts
• Familiarity with PySpark and DataFrame API
• Understanding of distributed computing concepts

Installation Guide

Installing GraphFrames

You can install GraphFrames using the following methods:

Method 1: Using pip

pip install graphframes

Method 2: Using Spark packages during runtime

When starting a PySpark session, include the GraphFrames package:

pyspark --packages graphframes:graphframes:0.8.2-spark3.0-s_2.12

Method 3: Adding to Spark configuration in code

from pyspark.sql import SparkSession
 
spark = SparkSession.builder \
    .appName("GraphFrames Example") \
    .config("spark.jars.packages", "graphframes:graphframes:0.8.2-spark3.0-s_2.12") \
    .getOrCreate()

GraphFrames vs GraphX

GraphX

• Spark’s original graph processing library
• Written in Scala, with limited Python support
• Based on RDD API
• Optimized for graph-parallel computation
• Steeper learning curve

GraphFrames

• Modern graph processing library for Spark
• First-class Python support
• Based on DataFrame API
• Seamless integration with Spark SQL and MLlib
• More accessible API and better performance
• Support for property graphs (attributes on vertices and edges)
• Better for interactive analysis and integration with ML pipelines

Creating GraphFrames

A GraphFrame consists of two primary components:

1. Vertices DataFrame (must have an “id” column)
2. Edges DataFrame (must have “src” and “dst” columns)

Basic Example

from pyspark.sql import SparkSession
from graphframes import GraphFrame
 
# Initialize Spark session
spark = SparkSession.builder \
    .appName("GraphFrames Example") \
    .config("spark.jars.packages", "graphframes:graphframes:0.8.2-spark3.0-s_2.12") \
    .getOrCreate()
 
# Create vertices DataFrame
vertices = spark.createDataFrame([
    ("1", "Alice", 34),
    ("2", "Bob", 36),
    ("3", "Charlie", 30),
    ("4", "David", 29),
    ("5", "Esther", 32),
    ("6", "Fanny", 36)
], ["id", "name", "age"])
 
# Create edges DataFrame
edges = spark.createDataFrame([
    ("1", "2", "friend"),
    ("1", "3", "friend"),
    ("2", "3", "follow"),
    ("3", "4", "follow"),
    ("4", "5", "follow"),
    ("5", "6", "follow"),
    ("6", "4", "follow")
], ["src", "dst", "relationship"])
 
# Create a GraphFrame
g = GraphFrame(vertices, edges)
 
# Display the graph
print("Vertices:")
g.vertices.show()
print("Edges:")
g.edges.show()

Creating from Existing DataFrames

You can also create GraphFrames from existing DataFrames by ensuring they have the required columns:

# Make sure your existing DataFrames have the required columns
# Vertices: must have an "id" column
# Edges: must have "src" and "dst" columns
 
# Example with renaming columns if needed
existing_user_data = spark.table("users")
v = existing_user_data.withColumnRenamed("user_id", "id")
 
existing_connections = spark.table("connections")
e = existing_connections \
    .withColumnRenamed("from_user", "src") \
    .withColumnRenamed("to_user", "dst")
 
g = GraphFrame(v, e)

Common Graph Functions

Graph Queries

Finding vertices and edges

# Get a specific vertex
g.vertices.filter("id = '1'").show()
 
# Get specific edges
g.edges.filter("relationship = 'follow'").show()

Basic graph statistics

# Count vertices and edges
print(f"Number of vertices: {g.vertices.count()}")
print(f"Number of edges: {g.edges.count()}")

Degree Calculations

inDegrees

Calculates the number of incoming edges for each vertex:

# Calculate in-degrees
in_degrees = g.inDegrees
print("In-degrees:")
in_degrees.sort("inDegree", ascending=False).show()

outDegrees

Calculates the number of outgoing edges for each vertex:

# Calculate out-degrees
out_degrees = g.outDegrees
print("Out-degrees:")
out_degrees.sort("outDegree", ascending=False).show()

degrees

Calculates the total degree (in + out) for each vertex:

# Calculate total degrees
degrees = g.degrees
print("Total degrees:")
degrees.sort("degree", ascending=False).show()

Graph Algorithms

PageRank

Implements the PageRank algorithm to measure vertex importance:

# Run PageRank algorithm
results = g.pageRank(resetProbability=0.15, tol=0.01, maxIter=10)
 
# Show vertices with their PageRank scores
print("PageRank results:")
results.vertices.select("id", "pagerank").sort("pagerank", ascending=False).show()
 
# Show edges with their weights
print("PageRank edge weights:")
results.edges.select("src", "dst", "weight").sort("weight", ascending=False).show()

Shortest Paths

Finds the shortest paths from a set of landmark vertices to all other vertices:

# Find shortest paths from landmarks
landmarks = ["1", "4"]
paths = g.shortestPaths(landmarks=landmarks)
print("Shortest paths:")
paths.select("id", "distances").show()

Connected Components

Finds connected components of the graph:

# Find connected components
components = g.connectedComponents()
print("Connected components:")
components.select("id", "component").sort("component").show()

Strongly Connected Components

Finds strongly connected components in a directed graph:

# Find strongly connected components
stronglyConnected = g.stronglyConnectedComponents(maxIter=10)
print("Strongly connected components:")
stronglyConnected.select("id", "component").sort("component").show()

Triangle Count

Counts the number of triangles each vertex is part of:

# Count triangles
triangleCounts = g.triangleCount()
print("Triangle counts:")
triangleCounts.select("id", "count").sort("count", ascending=False).show()

Label Propagation

Implements community detection using label propagation:

# Run label propagation algorithm
communities = g.labelPropagation(maxIter=5)
print("Communities from label propagation:")
communities.select("id", "label").sort("label").show()

Advanced GraphFrames Features

Motif Finding

Finds patterns in the graph using Spark’s graph pattern matching:

# Find a path of length 2 (A→B→C)
motifs = g.find("(a)-[e1]->(b); (b)-[e2]->(c)")
print("Paths of length 2:")
motifs.select("a.id", "b.id", "c.id").show()
 
# Find triangles
triangles = g.find("(a)-[e1]->(b); (b)-[e2]->(c); (c)-[e3]->(a)")
print("Triangles:")
triangles.select("a.id", "b.id", "c.id").show()

Subgraphs

Creates a subgraph by filtering vertices and edges:

# Create a subgraph of users older than 30 and 'friend' relationships
subgraph = g.filterVertices("age > 30").filterEdges("relationship = 'friend'")
print("Subgraph vertices:")
subgraph.vertices.show()
print("Subgraph edges:")
subgraph.edges.show()

BFS (Breadth-First Search)

Performs a breadth-first search from a given source vertex:

# Run BFS from vertex 1 to all reachable vertices
paths = g.bfs("id = '1'", "id = '6'")
print("BFS paths from 1 to 6:")
paths.show()

Complete Example with Analysis

Here’s a comprehensive example that demonstrates several GraphFrames features:

from pyspark.sql import SparkSession
from graphframes import GraphFrame
from pyspark.sql.functions import desc
 
# Initialize Spark session
spark = SparkSession.builder \
    .appName("GraphFrames Complete Example") \
    .config("spark.jars.packages", "graphframes:graphframes:0.8.2-spark3.0-s_2.12") \
    .getOrCreate()
 
# Create a more complex social network graph
vertices = spark.createDataFrame([
    ("1", "Alice", "NY", 34),
    ("2", "Bob", "CA", 36),
    ("3", "Charlie", "NY", 30),
    ("4", "David", "TX", 29),
    ("5", "Esther", "OH", 32),
    ("6", "Fanny", "CA", 36),
    ("7", "Gabby", "NY", 60),
    ("8", "Harry", "TX", 25)
], ["id", "name", "state", "age"])
 
edges = spark.createDataFrame([
    ("1", "2", "friend", 5),
    ("1", "3", "friend", 3),
    ("2", "3", "friend", 8),
    ("3", "4", "follow", 2),
    ("4", "5", "follow", 1),
    ("5", "6", "follow", 3),
    ("6", "4", "follow", 4),
    ("7", "1", "follow", 2),
    ("8", "2", "friend", 7),
    ("8", "3", "follow", 9),
    ("8", "4", "follow", 5)
], ["src", "dst", "relationship", "weight"])
 
# Create the GraphFrame
g = GraphFrame(vertices, edges)
 
print("====== Graph Analysis ======")
 
# Basic statistics
print(f"Number of vertices: {g.vertices.count()}")
print(f"Number of edges: {g.edges.count()}")
 
# Degree distributions
print("\n=== Degree Analysis ===")
g.inDegrees.groupBy("inDegree").count().orderBy("inDegree").show()
g.outDegrees.groupBy("outDegree").count().orderBy("outDegree").show()
 
# Find important vertices using PageRank
print("\n=== PageRank Analysis ===")
pagerank = g.pageRank(resetProbability=0.15, tol=0.01, maxIter=10)
important_vertices = pagerank.vertices.select("id", "name", "pagerank").orderBy(desc("pagerank"))
important_vertices.show()
 
# Community detection
print("\n=== Community Detection ===")
communities = g.labelPropagation(maxIter=5)
community_counts = communities.groupBy("label").count().orderBy(desc("count"))
print("Communities and their sizes:")
community_counts.show()
 
# Join community labels with original vertices
vertex_communities = communities.join(vertices, "id").select("id", "name", "state", "label")
print("Vertices and their communities:")
vertex_communities.orderBy("label", "id").show()
 
# Find patterns: People who are friends and from the same state
print("\n=== Pattern Matching: Friends from same state ===")
friends_same_state = g.find("(a)-[e]->(b)") \
    .filter("e.relationship = 'friend'") \
    .filter("a.state = b.state")
friends_same_state.select("a.name", "b.name", "a.state").show()
 
# Shortest paths analysis
print("\n=== Shortest Paths Analysis ===")
landmarks = ["1", "8"]  # Find paths from Alice and Harry
paths = g.shortestPaths(landmarks=landmarks)
paths.select("id", "name", "distances").show()
 
# Triangle counting
print("\n=== Triangle Analysis ===")
triangles = g.triangleCount()
triangles.select("id", "name", "count").orderBy(desc("count")).show()
 
# Find influential followers (high pagerank with many outgoing edges)
print("\n=== Influential Followers Analysis ===")
influential = pagerank.vertices \
    .join(g.outDegrees, "id") \
    .select("id", "name", "pagerank", "outDegree") \
    .filter("relationship = 'follow'") \
    .orderBy(desc("pagerank"))
influential.show()
 
# Graph motifs: Find potential friendships (friend of a friend)
print("\n=== Friend Recommendations ===")
friend_of_friend = g.find("(a)-[e1]->(b); (b)-[e2]->(c)") \
    .filter("e1.relationship = 'friend'") \
    .filter("e2.relationship = 'friend'") \
    .filter("a.id != c.id") \
    .filter("NOT EXISTS (" +
            "SELECT * FROM edges " +
            "WHERE edges.src = a.id AND edges.dst = c.id" +
            ")")
friend_recommendations = friend_of_friend.select("a.id", "a.name", "c.id", "c.name").distinct()
print("People who should be friends (friend of friend but not currently friends):")
friend_recommendations.show()

Using GraphFrames in Jupyter Notebook

Setup in Jupyter

To use GraphFrames in Jupyter Notebook, you need to configure Spark properly:

# In the first cell of your Jupyter notebook:
import os
import sys
 
# Set environment variables if needed
os.environ['PYSPARK_PYTHON'] = sys.executable
os.environ['PYSPARK_DRIVER_PYTHON'] = sys.executable
 
# Initialize SparkSession with GraphFrames
from pyspark.sql import SparkSession
 
spark = SparkSession.builder \
    .appName("GraphFrames Jupyter") \
    .config("spark.jars.packages", "graphframes:graphframes:0.8.2-spark3.0-s_2.12") \
    .getOrCreate()
 
# Import GraphFrames
from graphframes import GraphFrame
 
# Now you can use GraphFrames

Visualizing Graphs in Jupyter

You can visualize GraphFrames in Jupyter using libraries like NetworkX and Matplotlib:

import networkx as nx
import matplotlib.pyplot as plt
from pyspark.sql.functions import col
 
# Convert GraphFrame to NetworkX graph
def graphframe_to_networkx(g):
    # Get vertices and edges as dictionaries
    vertices = {row['id']: row.asDict() for row in g.vertices.collect()}
    edges = [(row['src'], row['dst']) for row in g.edges.collect()]
    
    # Create NetworkX graph
    G = nx.DiGraph()
    
    # Add nodes with attributes
    for node_id, attrs in vertices.items():
        G.add_node(node_id, **{k: v for k, v in attrs.items() if k != 'id'})
    
    # Add edges
    G.add_edges_from(edges)
    
    return G
 
# Create NetworkX graph from GraphFrame
G = graphframe_to_networkx(g)
 
# Visualize the graph
plt.figure(figsize=(12, 8))
pos = nx.spring_layout(G)
nx.draw(G, pos, with_labels=True, node_color='skyblue', node_size=1500, 
        edge_color='gray', arrowsize=20, font_size=15)
plt.title("Graph Visualization", fontsize=20)
plt.show()
 
# Visualize with PageRank information
pagerank = g.pageRank(resetProbability=0.15, maxIter=10)
pr_values = {row['id']: row['pagerank'] for row in pagerank.vertices.collect()}
 
# Update node sizes based on PageRank
node_sizes = [pr_values[node] * 10000 for node in G.nodes()]
 
plt.figure(figsize=(12, 8))
nx.draw(G, pos, with_labels=True, node_color='lightgreen', 
        node_size=node_sizes, edge_color='gray', arrowsize=20, font_size=15)
plt.title("Graph with PageRank (node size)", fontsize=20)
plt.show()

Interactive Dashboard in Jupyter

For a more interactive experience, you can use libraries like Plotly:

import plotly.graph_objects as go
import networkx as nx
import pandas as pd
 
# Convert GraphFrame to NetworkX
G = graphframe_to_networkx(g)
 
# Get positions for nodes
pos = nx.spring_layout(G)
 
# Create edge traces
edge_x = []
edge_y = []
for edge in G.edges():
    x0, y0 = pos[edge[0]]
    x1, y1 = pos[edge[1]]
    edge_x.extend([x0, x1, None])
    edge_y.extend([y0, y1, None])
 
edge_trace = go.Scatter(
    x=edge_x, y=edge_y,
    line=dict(width=0.8, color='#888'),
    hoverinfo='none',
    mode='lines')
 
# Create node traces
node_x = []
node_y = []
for node in G.nodes():
    x, y = pos[node]
    node_x.append(x)
    node_y.append(y)
 
node_trace = go.Scatter(
    x=node_x, y=node_y,
    mode='markers+text',
    text=list(G.nodes()),
    textposition="top center",
    hoverinfo='text',
    marker=dict(
        showscale=True,
        colorscale='YlGnBu',
        size=20,
        colorbar=dict(
            thickness=15,
            title='PageRank',
            xanchor='left',
            titleside='right'
        ),
        line_width=2))
 
# Add node attributes
node_text = []
node_colors = []
for node in G.nodes():
    node_info = f"ID: {node}<br>"
    for attr, value in G.nodes[node].items():
        node_info += f"{attr}: {value}<br>"
    node_text.append(node_info)
    # Use PageRank for color intensity
    node_colors.append(pr_values[node])
 
node_trace.marker.color = node_colors
node_trace.hovertext = node_text
 
# Create figure
fig = go.Figure(data=[edge_trace, node_trace],
                layout=go.Layout(
                    title='Interactive Graph Visualization',
                    titlefont_size=16,
                    showlegend=False,
                    hovermode='closest',
                    margin=dict(b=20,l=5,r=5,t=40),
                    xaxis=dict(showgrid=False, zeroline=False, showticklabels=False),
                    yaxis=dict(showgrid=False, zeroline=False, showticklabels=False))
                )
 
fig.show()

Troubleshooting and Common Issues

Package Not Found

If you encounter a ClassNotFoundException or similar:

# Make sure you're using the correct package version for your Spark version
spark = SparkSession.builder \
    .config("spark.jars.packages", "graphframes:graphframes:0.8.2-spark3.0-s_2.12") \
    .getOrCreate()

Memory Issues

For large graphs, you might need to increase Spark’s memory allocation:

spark = SparkSession.builder \
    .config("spark.driver.memory", "4g") \
    .config("spark.executor.memory", "4g") \
    .config("spark.jars.packages", "graphframes:graphframes:0.8.2-spark3.0-s_2.12") \
    .getOrCreate()

Algorithm Not Converging

For iterative algorithms (PageRank, connected components), you might need to increase iterations:

# Increase maximum iterations
results = g.pageRank(resetProbability=0.15, maxIter=50, tol=0.001)

Summary

GraphFrames is a powerful library that brings graph processing capabilities to Apache Spark, leveraging the DataFrame API for ease of use and performance. It provides a wide range of graph algorithms and operations that can be used for network analysis, recommendation systems, path finding, community detection, and more.

The main advantages of GraphFrames over GraphX include:

1. First-class Python support with a clean API
2. Integration with Spark’s DataFrame API
3. Support for property graphs with rich attributes
4. Better performance for many operations
5. Easier integration with Spark ML pipelines

With GraphFrames, you can analyze complex graphs at scale using the familiar Spark programming model, making it an excellent choice for large-scale graph analytics.