如何在 Spark 1.3 中通过 Hive 指示 SparkSQL 中的数据库

Posted 2023-04-18

【中文标题】如何在 Spark 1.3 中通过 Hive 指示 SparkSQL 中的数据库

【英文标题】：How to indicate the database in SparkSQL over Hive in Spark 1.3

【发布时间】：2016-06-21 18:46:20

【问题描述】：

我有一个简单的 Scala 代码，它从 Hive 数据库中检索数据并从结果集中创建一个 RDD。它适用于 HiveContext。代码类似这样：

val hc = new HiveContext(sc)
val mysql = "select PRODUCT_CODE, DATA_UNIT from account"
hc.sql("use myDatabase")
val rdd = hc.sql(mySql).rdd

我使用的 Spark 版本是 1.3。问题是 hive.execution.engine 的默认设置是“mr”，这使得 Hive 使用很慢的 MapReduce。不幸的是，我不能强迫它使用“火花”。 我尝试通过替换 hc = new SQLContext(sc) 来使用 SQLContext 来查看性能是否会提高。有了这个改变行

hc.sql("use myDatabase")

正在抛出以下异常：

Exception in thread "main" java.lang.RuntimeException: [1.1] failure: ``insert'' expected but identifier use found

use myDatabase
^

Spark 1.3 文档说 SparkSQL 可以使用 Hive 表。我的问题是如何表明我想使用某个数据库而不是默认数据库。

【问题讨论】：

您是否尝试过常规的 Hive 语法，即select * from mydb.mytable？

是 - 出现另一个错误：java.lang.RuntimeException: Table Not Found: myDatabase.account

【参考方案1】：

使用数据库

在更高版本的 Spark 中支持

https://docs.databricks.com/spark/latest/spark-sql/language-manual/use-database.html

您需要将语句放在两个单独的 spark.sql 调用中，如下所示：

spark.sql("use mydb")
spark.sql("select * from mytab_in_mydb").show

【参考方案2】：

返回创建 HiveContext。 Hive 上下文使您能够使用 Hive 的元存储创建数据框。 Spark 仅使用 hive 中的元存储，不使用 hive 作为处理引擎来检索数据。因此，当您使用 sql 查询创建 df 时，它实际上只是询问 hive 的元存储“数据在哪里，数据的格式是什么”

Spark 获取该信息，并将针对 HDFS 上的基础数据运行进程。所以 Spark 正在执行查询，而不是 hive。

当您创建 sqlContext 时，它会删除 Spark 和 Hive 元存储之间的链接，因此错误是说它不了解您想要做什么。

【讨论】：

谢谢你，乔，你的评论。我正在使用 HiveContext 并注意到查询对大量数据运行非常缓慢。我正在使用 Spark 1.3，无法控制升级到 1.5+。不幸的是，在我的环境中，“hive.execution.engine”设置为“mr”，这使得 Hive 的运行速度比设置为“spark”时慢得多。所以我实际上想通过在运行查询中使用 sqlContext 而不是 MapReduce 来为 Spark 提供更多控制。奇怪的是，虽然 Spark 声称支持 Hive，但它并不理解它的数据库部分，它只是默认数据库的一个子目录。

Hive 没有做任何处理，所以 MR 没有做任何处理。 Spark 确实了解其中的数据库部分，它从元存储中获取所有这些信息。 SparkSQL 然后处理来自 HDFS 的数据。

Joe，抱歉，我不明白它是如何回答我的问题的

【参考方案3】：

我无法实现 use databale 命令，但这里有一个使用所需数据库的解决方法：

spark-shell --queue QUEUENAME; val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc) val res2 = sqlContext.sql("select count(1) from DB_NAME.TABLE_NAME") res2.collect()