package com.abaghel.examples.spark.graphframe; import java.util.ArrayList; import java.util.List; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; import org.apache.spark.sql.SparkSession; import org.graphframes.GraphFrame; import org.graphframes.lib.PageRank; /** * Sample application shows how to create a GraphFrame, query it, and run the PageRank algorithm. * * @author abaghel * */ public class SparkGraphFrameSample { public static void main(String[] args) { SparkSession spark = SparkSession.builder() .appName("SparkGraphFrameSample") .config("spark.sql.warehouse.dir", "/file:C:/temp") .master("local[2]") .getOrCreate(); //Create a Vertex DataFrame with unique ID column "id" List uList = new ArrayList() { { add(new User("a", "Alice", 34)); add(new User("b", "Bob", 36)); add(new User("c", "Charlie", 30)); } }; Dataset verDF = spark.createDataFrame(uList, User.class); //Create an Edge DataFrame with "src" and "dst" columns List rList = new ArrayList() { { add(new Relation("a", "b", "friend")); add(new Relation("b", "c", "follow")); add(new Relation("c", "b", "follow")); } }; Dataset edgDF = spark.createDataFrame(rList, Relation.class); //Create a GraphFrame GraphFrame gFrame = new GraphFrame(verDF, edgDF); //Get in-degree of each vertex. gFrame.inDegrees().show(); //Count the number of "follow" connections in the graph. long count = gFrame.edges().filter("relationship = 'follow'").count(); //Run PageRank algorithm, and show results. PageRank pRank = gFrame.pageRank().resetProbability(0.01).maxIter(5); pRank.run().vertices().select("id", "pagerank").show(); //stop spark.stop(); } }
User.java
package com.abaghel.examples.spark.graphframe; /** * User class * * @author abaghel * */ public class User { private String id; private String name; private int age; public User(){ } public User(String id, String name, int age) { super(); this.id = id; this.name = name; this.age = age; } public String getId() { return id; } public void setId(String id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } public int getAge() { return age; } public void setAge(int age) { this.age = age; } }
Relation.java
package com.abaghel.examples.spark.graphframe; /** * Relation class * * @author abaghel * */ public class Relation { private String src; private String dst; private String relationship; public Relation(){ } public Relation(String src, String dst, String relationship) { super(); this.src = src; this.dst = dst; this.relationship = relationship; } public String getSrc() { return src; } public void setSrc(String src) { this.src = src; } public String getDst() { return dst; } public void setDst(String dst) { this.dst = dst; } public String getRelationship() { return relationship; } public void setRelationship(String relationship) { this.relationship = relationship; } }
控制台输出
16/08/27 22:34:45 INFO DAGScheduler: Job 10 finished: show at SparkGraphFrameSample.java:56, took 0.938910 s 16/08/27 22:34:45 INFO CodeGenerator: Code generated in 6.599005 ms +---+-------------------+ | id| pagerank| +---+-------------------+ | a| 0.01| | b|0.08763274109799998| | c| 0.077926810699| +---+-------------------+