java - org.apache.spark.sql.AnalysisException : Table and view not found

Question

我正在使用neo4j-spark connector将neo4j数据提取到spark数据帧中。我能够成功获取它，因为我能够显示数据框。然后我用 createOrReplaceTempView() 注册数据帧方法。然后我尝试在其上运行 Spark sql，但它给出了异常:

org.apache.spark.sql.AnalysisException: Table or view not found: neo4jtable;

这就是我的整个代码的样子:

import java.text.ParseException;

import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.AnalysisException;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.neo4j.spark.Neo4JavaSparkContext;
import org.neo4j.spark.Neo4j;

import scala.collection.immutable.HashMap;

public class Neo4jDF {

    private static Neo4JavaSparkContext neo4jjsc;
    private static SparkConf sConf;
    private static JavaSparkContext jsc;
    private static SparkContext sc; 
    private static SparkSession ss;

    private static Dataset<Row> neo4jdf;
    static String neo4jip = "ll.mm.nn.oo";

    public static void main(String[] args) throws AnalysisException, ParseException 
    {
        setSparkConf();
        setJavaSparkContext();
        setNeo4jJavaSparkContext();
        setSparkContext();
        setSparkSession();

        neo4jdf = loadNeo4jDataframe();
        neo4jdf.createOrReplaceTempView("neo4jtable");

        neo4jdf.show(false); //this prints correctly

        Dataset<Row> neo4jdfsqled = ss.sql("SELECT * from neo4jtable");

        neo4jdfsqled.show(false); //this throws exception
    }

    private static Dataset<Row> loadNeo4jDataframe(String pAutosysBoxCaption)
    {
        Neo4j neo4j = new Neo4j(jsc.sc());
        HashMap<String, Object> a = new HashMap<String, Object>();
        Dataset<Row> rdd = neo4j.cypher("cypher query deleted for irrelevance", a).loadDataFrame();
        return rdd;
    }

    private static void setSparkConf()
    {
        sConf = new SparkConf().setAppName("GetNeo4jToRddDemo");
        sConf.set("spark.neo4j.bolt.url", "bolt://" + neo4jip + ":7687");
        sConf.set("spark.neo4j.bolt.user", "neo4j");
        sConf.set("spark.neo4j.bolt.password", "admin");
        sConf.setMaster("local");
        sConf.set("spark.testing.memory", "471859200");
        sConf.set("spark.sql.warehouse.dir", "file:///D:/Mahesh/workspaces/spark-warehouse");
    }

    private static void setJavaSparkContext()
    {
        jsc = new JavaSparkContext(sConf);
    }

    private static void setSparkContext()
    {
        sc = JavaSparkContext.toSparkContext(jsc);
    }

    private static void setSparkSession()
    {
        ss = new SparkSession(sc);
    }

    private static void setNeo4jJavaSparkContext()
    {
        neo4jjsc = Neo4JavaSparkContext.neo4jContext(jsc);
    }
}

我觉得问题可能在于如何创建所有 Spark 环境变量。我首先创建了SparkConf sConf .
来自 sConf ，我创建了JavaSparkContext jsc .
来自 jsc ，我创建了SparkContext sc .
来自 sc ，我创建了SparkSession ss .
来自 ss ，我创建了 Neo4jJavaSparkContext neo4jjjsc .

从视觉上看:

sConf -> jsc -> sc       -> ss 
             -> neo4jjsc 

另请注意

• 在 loadNeo4jDataframe() 内部，我使用 sc实例化实例 Neo4j neo4j ，然后用于获取 Neo4j 数据。
• 使用 Neo4j 获取数据实例。
• neo4jjjsc从未使用过，但我将其保留为可能的问题提示。

鉴于所有这些要点和观察结果，请告诉我为什么我会出现表未找到异常？我一定是错过了一些愚蠢的事情。 :\

更新

尝试设置ss (使用neo4j的SparkContext获取数据后)如下:

private static void setSparkSession(SparkContext sc)
{
    ss = new SparkSession(sc);
}

private static Dataset<Row> loadNeo4jDataframe(String pAutosysBoxCaption)
{
    Neo4j neo4j = new Neo4j(sc);

    Dataset<Row> rdd = neo4j.cypher("deleted cypher for irrelevance", a).loadDataFrame();

    //initalizing ss after data is fetched using SparkContext of neo4j
    setSparkSession(neo4j.sc());  
    return rdd;
}

更新2

从评论中，刚刚意识到neo4j使用 Spark 上下文创建自己的 Spark session sc提供给它的实例。我无法访问该 Spark session 。那么，我应该如何将在其他 Spark session (此处由 neo4jdf 创建的 Spark session )中创建的任意数据帧(此处为 neo4j.cypher )添加/注册到我的 Spark session ss 中？

Answer 1

根据症状，我们可以推断这两段代码使用了不同的 SparkSession/SQLContext。假设 Neo4j 连接器中没有发生任何异常情况，您应该能够通过更改来解决此问题:

private static void setSparkSession()
{
    ss = SparkSession().builder.getOrCreate();
}

或者在调用 setNeo4jJavaSparkContext 之前初始化 SparkSession。

如果这些不起作用，您可以改用 createGlobalTempView。

重要:

一般来说，我建议使用构建器模式初始化单个 SparkSession，并在必要时从中派生其他上下文 (SparkContexts)。