SparkSession.sql 和 Dataset.sqlContext.sql 有啥区别？

Posted 2023-04-15

【中文标题】SparkSession.sql 和 Dataset.sqlContext.sql 有啥区别？

【英文标题】：What's the difference between SparkSession.sql and Dataset.sqlContext.sql?SparkSession.sql 和 Dataset.sqlContext.sql 有什么区别？

【发布时间】：2017-12-07 09:06:55

【问题描述】：

我有以下代码的sn-ps，我想知道这两者之间有什么区别，我应该使用哪一个？我正在使用火花 2.2。

Dataset<Row> df = sparkSession.readStream()
    .format("kafka")
    .load();

df.createOrReplaceTempView("table");
df.printSchema();

Dataset<Row> resultSet =  df.sqlContext().sql("select value from table"); //sparkSession.sql(this.query);
StreamingQuery streamingQuery = resultSet
        .writeStream()
        .trigger(Trigger.ProcessingTime(1000))
        .format("console")
        .start();

对

Dataset<Row> df = sparkSession.readStream()
    .format("kafka")
    .load();

df.createOrReplaceTempView("table");

Dataset<Row> resultSet =  sparkSession.sql("select value from table"); //sparkSession.sql(this.query);
StreamingQuery streamingQuery = resultSet
        .writeStream()
        .trigger(Trigger.ProcessingTime(1000))
        .format("console")
        .start();

【参考方案1】：

sparkSession.sql("sql query") 与 df.sqlContext().sql("sql query") 之间存在非常细微的区别。

请注意，您可以在单个 Spark 应用程序中拥有零个、两个或多个 SparkSessions（但假设您在 Spark SQL 中至少且通常只有一个 SparkSession 应用程序）。

还请注意，Dataset 绑定到在其中创建它的 SparkSession，SparkSession 永远不会改变。

您可能想知道为什么有人会想要它，但这为您提供了查询之间的界限，您可以为不同的数据集使用相同的表名，这实际上是 Spark SQL 的一个非常强大的功能。

以下示例显示了差异，希望能让您了解它为何如此强大。

scala> spark.version
res0: String = 2.3.0-SNAPSHOT

scala> :type spark
org.apache.spark.sql.SparkSession

scala> spark.sql("show tables").show
+--------+---------+-----------+
|database|tableName|isTemporary|
+--------+---------+-----------+
+--------+---------+-----------+

scala> val df = spark.range(5)
df: org.apache.spark.sql.Dataset[Long] = [id: bigint]

scala> df.sqlContext.sql("show tables").show
+--------+---------+-----------+
|database|tableName|isTemporary|
+--------+---------+-----------+
+--------+---------+-----------+

scala> val anotherSession = spark.newSession
anotherSession: org.apache.spark.sql.SparkSession = org.apache.spark.sql.SparkSession@195c5803

scala> anotherSession.range(10).createOrReplaceTempView("new_table")

scala> anotherSession.sql("show tables").show
+--------+---------+-----------+
|database|tableName|isTemporary|
+--------+---------+-----------+
|        |new_table|       true|
+--------+---------+-----------+


scala> df.sqlContext.sql("show tables").show
+--------+---------+-----------+
|database|tableName|isTemporary|
+--------+---------+-----------+
+--------+---------+-----------+