如何在 Java 中的 SparkSQL 中正确创建视图

Posted 2023-04-15

【中文标题】如何在 Java 中的 SparkSQL 中正确创建视图

【英文标题】：How to correctly create a view in SparkSQL in Java

【发布时间】：2021-05-30 13:01:11

【问题描述】：

我正在尝试做一件非常基本的事情，但我遇到了一个我找不到解决方案的错误。

我想要做的是能够在数据帧上执行 SQL 查询。为此，我在文档here 中看到，您需要从数据框创建一个视图。问题是，虽然我完全按照文档中的内容进行操作，但 IntelliJ 一直给我一个编译错误。

这是我的代码：

    @Override
public Long execute() 
    log.info("Starting processing query");
    Instant start = Instant.now();


    Dataset<Row> dataframe = this.hdfsIO.readParquetAsDataframe(vaccineAdministrationSummaryFile);
//    Dataset<Row> dataframe = this.sparkSession.read().parquet(this.hdfsUrl + inputDir + "/" + filename);
    dataframe.createOrReplaceTempView("query");

    Dataset<Row> sqlDF = sparkSession.sql("SELECT * FROM query");

sparkSession.sql(SELECT * FROM query); 行给了我一个错误和警告：

警告：未配置任何数据源来运行此 SQL 并提供高级代码帮助。通过问题菜单禁用此检查（alt+enter）。 错误：无法解析表“查询”。

如果我的代码与文档中的代码几乎相同，为什么会发生这种情况？

在dataframe.createOrReplaceTempView("query"); 之后运行sparkSession.catalog().listTables().show(false); 我明白了：

21/05/31 09:36:29 INFO CodeGenerator: Code generated in 27.591936 ms
21/05/31 09:36:29 INFO CodeGenerator: Code generated in 11.52196 ms
21/05/31 09:36:29 INFO CodeGenerator: Code generated in 15.643281 ms
+-----+--------+-----------+---------+-----------+
|name |database|description|tableType|isTemporary|
+-----+--------+-----------+---------+-----------+
|query|null    |null       |TEMPORARY|true       |
+-----+--------+-----------+---------+-----------+

相反，在dataframe.createOrReplaceTempView("query"); 之前运行相同的命令我明白了：

+----+--------+-----------+---------+-----------+
|name|database|description|tableType|isTemporary|
+----+--------+-----------+---------+-----------+
+----+--------+-----------+---------+-----------+

函数readParquetAsDataframe()定义如下，传递的字符串是hdfs中文件的路径：

    public Dataset<Row> readParquetAsDataframe(String filename)
        return this.sparkSession.read().parquet(this.hdfsUrl + inputDir + "/" + filename);

我还认为问题可能在于我如何构建 SparkSession，因为我没有指定任何配置，但如果是这样，我什至不知道如何正确配置它。 SparkSession 的构建如下：

SparkSession spark = SparkSession
    .builder()
    .appName("Project 1")
    .getOrCreate();

【问题讨论】：

这能回答你的问题吗？ How does createOrReplaceTempView work in Spark?

不，我已经阅读了那个答案，但它包含文档中已经提供的信息，仅此而已。

你真的需要原始sql吗？使用可用的数据框选择方法在编译时更加优化

老实说，我想我可以不用它，但我试图写这个指令，因为我得到了这个错误，我不明白为什么我发布这个问题来了解正在发生的事情..跨度>

如果您将spark.catalog.listTables.show(false) 放在createOrReplaceTempView(...) 之后，您会看到什么？

【参考方案1】：

看起来你有 java 代码。查看此 JDBC 示例，收集 DataSet<Row> 后阅读 parquet 没有什么不同，请注意 cmets。

public static void main(String[] args) throws Exception 
        
        SparkSession sparkSession = SparkSession.builder()
                                .master("local")
                                .appName("Test App")
                                .getOrCreate();

        // JDBC connection details
        String driver = "com.mysql.cj.jdbc.Driver";
        String url = "jdbc:mysql://192.168.1.113:3306/db";
        String user = "user";
        String pass = "password";
        
        // JDBC Connection and load table in Dataframe
        Dataset<Row> trans = sparkSession.read()
                                .format("jdbc")
                                .option("driver", driver)
                                .option("url", url)
                                //your table name here
                                .option("dbtable", "<table name>")
                                .option("user", user)
                                .option("password", pass).load();
        
        //your table view name here
        trans.createOrReplaceTempView("<table_view_name>");

        Dataset<Row> sqlResult = sparkSession
                                    //your same table view name here
                                    .sql("select firstname,lastname from <table_view_name>");

        sqlResult.foreach(f -> 
            //print the firstname from your table
            System.out.println("firstname -> "+f.get(0));
        );