Как создать простой искровой граф с помощью java?

По сути, я разработчик Java, и теперь у меня есть возможность поработать над Spark, и я изучил основы API Spark, например, что такое SparkConfig, SparkContaxt, RDD, SQLContaxt, DataFrame, DataSet, а затем я могу выполнить несколько простых простых преобразований, используя RDD, SQL .... но когда я пытаюсь отработать какое-то приложение графического фрейма с использованием java, я не могу добиться успеха, и я просмотрел так много руководств по YouTube, форумов и потоков stackoverflow, но нигде я не нашел прямого предложения или решение. На самом деле я сталкиваюсь с этой проблемой, когда пытаюсь создать объект для класса GraphFrame, и я также загрузил рецептивную банку (graphframes-0.2.0-spark2.0-s_2.11.jar), но все еще сталкиваюсь с проблемой сейчас, я хочу довести свой анализ до того, чего я достиг, из-за того, что я очень новичок в Spark, я не могу двигаться дальше, поэтому, если кто-то поможет мне, это действительно полезно для всех. Заранее спасибо. Исключением является то, что я сталкиваюсь с конструктором GraphFrame(DataFrame, DataFrame) undefined

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.storage.StorageLevel;
import org.graphframes.GraphFrame;

import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.databind.JsonMappingException;

public class SparkJavaGraphFrameOne {

    public static void main(String[] args) throws JsonParseException, JsonMappingException, IOException{

        SparkConf conf = new SparkConf().setAppName("test").setMaster("local");

        JavaSparkContext sc = new JavaSparkContext(conf);
        SQLContext sqlContext = new org.apache.spark.sql.SQLContext(sc);

        JavaRDD<Row> verRow = sc.parallelize(Arrays.asList(RowFactory.create(1,"A"),RowFactory.create(2,"B")));
        JavaRDD<Row> edgRow = sc.parallelize(Arrays.asList(RowFactory.create(1,2,"Edge")));     

        List<StructField> verFields = new ArrayList<StructField>();
        verFields.add(DataTypes.createStructField("id",DataTypes.IntegerType, true));
        verFields.add(DataTypes.createStructField("name",DataTypes.StringType, true));

        List<StructField> EdgFields = new ArrayList<StructField>();
        EdgFields.add(DataTypes.createStructField("fromId",DataTypes.IntegerType, true));
        EdgFields.add(DataTypes.createStructField("toId",DataTypes.IntegerType, true));
        EdgFields.add(DataTypes.createStructField("name",DataTypes.StringType, true));

        StructType verSchema = DataTypes.createStructType(verFields);
        StructType edgSchema = DataTypes.createStructType(EdgFields);

        DataFrame verDF = sqlContext.createDataFrame(verRow, verSchema);
        DataFrame edgDF = sqlContext.createDataFrame(edgRow, edgSchema);

        GraphFrame g = new GraphFrame(verDF,edgDF);
        g.vertices().show();
        g.edges().show();
        g.persist(StorageLevel.MEMORY_AND_DISK());
    }

}

person Venkaiah Yepuri    schedule 26.08.2016    source источник


Ответы (4)


Я написал пример программы на java, используя Spark 2.0.0 и GraphFrame 0.2.0. Эта программа основана на образце программы, представленном на http://graphframes.github.io/quick-start.html#start-using-graphframes. Надеюсь это поможет.

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.abaghel.examples.spark</groupId>
<artifactId>spark-graphframe</artifactId>
<version>1.0.0-SNAPSHOT</version>

<dependencies>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.11</artifactId>
        <version>2.0.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-graphx_2.11</artifactId>
        <version>2.0.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.11</artifactId>
        <version>2.0.0</version>
    </dependency>
    <dependency>
        <groupId>graphframes</groupId>
        <artifactId>graphframes</artifactId>
        <version>0.2.0-spark2.0-s_2.11</version>
    </dependency>
</dependencies>

<repositories>
    <!-- list of other repositories -->
    <repository>
        <id>SparkPackagesRepo</id>
        <url>http://dl.bintray.com/spark-packages/maven</url>
    </repository>
 </repositories>
 <build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.1</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
    </plugins>
  </build>
</project>

SparkGraphFrameSample.java

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<User> uList = new ArrayList<User>() {
        {
            add(new User("a", "Alice", 34));
            add(new User("b", "Bob", 36));
            add(new User("c", "Charlie", 30));
        }
    };

    Dataset<Row> verDF = spark.createDataFrame(uList, User.class);

    //Create an Edge DataFrame with "src" and "dst" columns
    List<Relation> rList = new ArrayList<Relation>() {
        {
            add(new Relation("a", "b", "friend"));
            add(new Relation("b", "c", "follow"));
            add(new Relation("c", "b", "follow"));
        }
    };

    Dataset<Row> 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|
+---+-------------------+
person abaghel    schedule 26.08.2016
comment
Спасибо большое агагель. Попробую это и дам вам знать к концу дня. Еще раз большое спасибо, надеюсь, это действительно поможет мне двигаться дальше. - person Venkaiah Yepuri; 26.08.2016
comment
Ваша первоначальная проблема с конструктором new GraphFrame(verDF,edgDF) будет решена с помощью файлов, которые я предоставил в этом посте. Я ожидаю, что ответ будет принят, если этот образец версии Java действительно вам помог. - person abaghel; 27.08.2016
comment
abaghel - определенно проголосую за, но я столкнулся с некоторой проблемой при загрузке этой зависимости (jar 0.2.0-spark2.0-s_2.11) не могу загрузить. Любая проблема с версией Java, которую я использую 1.7. ? - person Venkaiah Yepuri; 27.08.2016
comment
Я использую Java 1.8. У меня не возникло проблем с загрузкой файла jar, поскольку зависимость для файла jar и URL-адрес репозитория определены в pom.xml. Я использую Eclipse Mars, и я построил с помощью команды установки maven, которая загрузила файлы jar. - person abaghel; 27.08.2016
comment
В любом случае, я загрузил эту банку извне и указал путь сборки. Теперь, когда я запускаю приложение, я получаю это исключение во время выполнения. Я сделал google, они предложили, что в файле SBT требуются некоторые изменения, но я не уверен, где находятся эти файлы. Исключением является исключение в потоке main java.lang.NoClassDefFoundError: com/typesafe/scalalogging/slf4j/LazyLogging. - person Venkaiah Yepuri; 27.08.2016
comment
abagel - можете опубликовать снимок экрана приложения. Показывает ли он какой-либо график или просто отображает отношения между объектами пользователей в виде текста, выводимого на вкладке консоли? - person Venkaiah Yepuri; 27.08.2016
comment
Вы используете сборку Scala и SBT? Я предлагаю вам взять образец кода и файл pom.xml, которые я разместил здесь, и запустить SparkGraphFrameSample. java в затмении. Затем вы можете добавить свой класс Java в проект и попытаться запустить его. Я добавил консольный вывод выше. - person abaghel; 27.08.2016
comment
Я сделал то же самое, но при запуске основного класса возникает это исключение. Пожалуйста, просмотрите этот текстовый файл для исключения полного стека мой полный код github.com/venkatrohith/sparkgraphframes - person Venkaiah Yepuri; 27.08.2016
comment
Венки, я клонировал ваш проект с github и построил его с помощью командной строки maven. Затем я запустил его с помощью команды mvn exec:java -Dexec.mainClass=com.spark.mygraphframe.SparkGraphFrameSample. Он работает нормально без каких-либо проблем, и я получаю результат. Я использую java 1.8. Попробуйте с java 1.8. Попробуйте очистить свою среду, попробуйте с другой машины и из командной строки maven. - person abaghel; 27.08.2016
comment
Давайте продолжим обсуждение в чате. - person Venkaiah Yepuri; 27.08.2016

Я не знаю, сможете ли вы решить свою проблему или нет. Я только что увидел вашу проблему. Я думаю, что для получения Exception в потоке "main" java.lang.NoClassDefFoundError: com/typesafe/scalalogging/slf4j/LazyLogging вам нужно поместить следующую банку в свой pom.xml

         <dependency>
            <groupId>com.typesafe.scala-logging</groupId>
             <artifactId>scala-logging-slf4j_2.10</artifactId>
            <version>2.1.2</version>
        </dependency>

Я столкнулся с той же проблемой, и, добавив эту банку, я смог решить эту проблему.

person Rhea    schedule 15.09.2016
comment
Я Трина, спасибо за ваше предложение. Но на самом деле эта проблема с версией Java после обновления версии Java до версии 8 затем решается. Надеюсь, вы узнали, кто я? - person Venkaiah Yepuri; 17.09.2016

Я могу воспроизвести проблему (работает непрерывно) в 0.5.0-spark2.1-s_2.11 и отлично работает в 0.4.0-spark2.1-s_2.11.

person Rajasekaran Kandhasamy    schedule 18.09.2017

Чтобы исправить проблему с конструктором GraphFrame, попробуйте:

GraphFrame gf = GraphFrame.apply(verDF, edgeDF);

person ilya1245    schedule 08.08.2019