使用用Pyspark和GraphX实现解析复杂网络数据

pip install pyspark

from pyspark.sql import SparkSession,Row
from datetime import datetime, date
import pandas as pd
import os
os.environ['PYSPARK_PYTHON'] = "%你的Python包路径%//python.exe"

spark = SparkSession.builder.getOrCreate()
df = spark.createDataFrame([
    Row(a=1, b=2., c='string1', d=date(2000, 1, 1), e=datetime(2000, 1, 1, 12, 0)),
    Row(a=2, b=3., c='string2', d=date(2000, 2, 1), e=datetime(2000, 1, 2, 12, 0)),
    Row(a=4, b=5., c='string3', d=date(2000, 3, 1), e=datetime(2000, 1, 3, 12, 0))
])
df.show()

SPARK_HOME  = C:\apps\opt\spark-3.5.0-bin-hadoop3
HADOOP_HOME = C:\apps\opt\spark-3.5.0-bin-hadoop3

pip install graphframes

from pyspark.sql.types import *
from pyspark.sql import SparkSession
from pyspark import SparkContext, SparkConf
import pandas as pd
from graphframes import GraphFrame

spark_conf = SparkConf().setAppName('Python_Spark_WordCount').setMaster('local[2]') 
sc = SparkContext(conf=spark_conf)
spark=SparkSession.builder.appName("graph").getOrCreate()
v = spark.createDataFrame([
  ("a", "Alice", 34),
  ("b", "Bob", 36),
  ("c", "Charlie", 30),
], ["id", "name", "age"])

# Create an Edge DataFrame with "src" and "dst" columns
e = spark.createDataFrame([
  ("a", "b", "friend"),
  ("b", "c", "follow"),
  ("c", "b", "follow"),
], ["src", "dst", "relationship"])
# Create a GraphFrame
g = GraphFrame(v, e)

# Query: Get in-degree of each vertex.
g.inDegrees.show()

# Query: Count the number of "follow" connections in the graph.
g.edges.filter("relationship = 'follow'").count()

# Run PageRank algorithm, and show results.
results = g.pageRank(resetProbability=0.01, maxIter=20)
results.vertices.select("id", "pagerank").show()

import os
os.environ['PYSPARK_PYTHON'] = "%你自己的Python路径%//Python//python.exe"

edges=sc.textFile(r'/Users/xiaoyu/edges')
edges=edges.map(lambda x:x.split('\t'))
edges_df=spark.createDataFrame(edges,['src','dst'])

nodes=sc.textFile(r'/Users/xiaoyu/nodes')
nodes=nodes.map(lambda x:[x]) 
nodes_df=spark.createDataFrame(nodes,['id'])
graph=GraphFrame(nodes_df, edges_df)

# 计算每个节点的入度和出度
in_degrees = graph.inDegrees
out_degrees = graph.outDegrees

# 打印节点的入度和出度
in_degrees.show()
out_degrees.show()

# 找到具有最大入度的节点
max_in_degree = in_degrees.agg(F.max("inDegree")).head()[0]
node_with_max_in_degree = in_degrees.filter(in_degrees.inDegree == max_in_degree).select("id")

# 找到具有最大出度的节点
max_out_degree = out_degrees.agg(F.max("outDegree")).head()[0]
node_with_max_out_degree = out_degrees.filter(out_degrees.outDegree == max_out_degree).select("id")

# 打印结果
node_with_max_in_degree.show()
node_with_max_out_degree.show()