使用 Spark 查询 hive 表

Posted 2023-04-17

【中文标题】使用 Spark 查询 hive 表

【英文标题】：Query hive table with Spark

【发布时间】：2018-03-02 13:42:57

【问题描述】：

我是 Apache Hive 和 Spark 的新手。我的 Hadoop 服务器上有一些现有的 Hive 表，我可以运行一些 HQL 命令并使用 hive 或直线从表中获取我想要的内容，例如，选择表的前 5 行。相反，我想使用 Spark 来实现相同的目标。我在服务器上的 Spark 版本是 1.6.3。

使用以下代码（我将我的数据库名称和表替换为 database 和 table）：

sc = SparkContext(conf = config)
sqlContext = HiveContext(sc)

query = sqlContext.createDataFrame(sqlContext.sql("SELECT * from database.table LIMIT 5").collect())
df = query.toPandas()
df.show()

我收到此错误：

ValueError: Some of types cannot be determined after inferring.  
Error:root: An unexpected error occurred while tokenizing input
The following traceback may be corrupted or invalid
The error message is: ('EOF in multi-line string', (1, 0))

但是，我可以将直线与相同的查询一起使用并查看结果。

经过一天的谷歌搜索和搜索，我将代码修改为：

table_ccx = sqlContext.table("database.table")
table_ccx.registerTemplate("temp")
sqlContext.sql("SELECT * FROM temp LIMIT 5").show()

现在错误消失了，但是除了一两个日期和列名之外，所有的行值都是空的。

我也试过

table_ccx.refreshTable("database.table")

它没有帮助。是否有需要我的 IT 团队执行的设置或配置？感谢您的帮助。

编辑：话虽如此，我的 python 代码适用于 Hadoop 上的某些表。不知道问题是因为表上的一些条目还是没有？如果是，那么相应的 beeline/Hive 命令是如何工作的？

【问题讨论】：

可能不相关，但sqlContext.createDataFrame(sqlContext.sql("SELECT * from database.table LIMIT 5").collect())真的没有必要。 sqlContext.sql("SELECT * from database.table LIMIT 5") 可以。

看起来表格内容不可读，可能是因为一些不匹配的配置。你能分享那个表的CREATE 语句吗？您可以通过运行SHOW CREATE TABLE database.table 来显示它。

你是对的。我不需要 sqlContext.createDataFrame。这个 sqlContext.sql("SELECT * from database.table LIMIT 5") 将解决这个问题。

我应该建议它作为答案吗？

query = sqlContext.sql("SELECT * from database.table LIMIT 5") 是 Spark DataFrame，无需从中创建 DataFrame。因此，如果我想要 Pandas DataFrame 出来，那么 query.toPandas() 将创建它。非常感谢@stefanobaghino

【参考方案1】：

当它出现在 cmets 中时，稍微整理一下代码就可以使事情正常进行。

问题出在这行代码上：

query = sqlContext.createDataFrame(sqlContext.sql("SELECT * from database.table LIMIT 5").collect())

你在这里做的是：

要求 Spark 查询数据源（这会创建一个 DataFrame） collect 驱动程序上的所有内容作为本地集合 将 Spark 上的本地集合与 createDataFrame 并行化

一般而言，该方法应该有效，尽管它显然是不必要的复杂。

以下会做：

query = sqlContext.sql("SELECT * from database.table LIMIT 5")

我不完全确定这件事为什么会破坏你的代码，但它仍然会破坏你的代码（正如它在 cmets 中出现的那样）并且它也改进了它。

【讨论】：

我按照您的建议更改了代码，但是，现在 Pandas DataFrame 中的所有条目都是 None ，我认为原因是 Spark DataFrame 架构认为该列应该是字符串并且我有整数所以它将所有内容更改为无。也许我需要以某种方式更改架构然后尝试它。我说的有道理吗？

query.schema --> 输出： StructType(List(StructField(*****,StringType, true))) 和 query.show(20) 将所有内容显示为空。因此，无论出于何种原因，它都会将所有内容更改为 null。

我想我找到了解决方案：无论出于何种原因，当我使用 Spark 2.1 时，条目都会被填充。