如何将DataFrame直接保存到Hive？

Question

是否可以将DataFrame火花直接保存到Hive中.

我已尝试转换DataFrame为Rdd然后保存为文本文件,然后加载到配置单元.但我想知道我是否可以直接保存dataframe到蜂巢

Answer 1

如果您正在使用saveAsTable(它更像是持久化您的数据帧),您必须确保为spark应用程序分配了足够的内存.对于大型数据集,您可以创建临时表并将其转储到配置单元表中.

您可以使用spark中可用的sqlContext对象.

让我们说你的数据框是myDf.您可以创建一个临时表.

myDf.createOrReplaceTempView("mytempTable") 

然后,您可以使用简单的hive语句来创建表并从临时表中转储数据.

sqlContext.sql("create table mytable as select * from mytempTable");

Answer 2

使用DataFrameWriter.saveAsTable.(df.write.saveAsTable(...))请参阅Spark SQL和DataFrame指南.

Answer 3

我不认为df.write.saveAsTable(...)Spark 2.0文档中已弃用.它在Amazon EMR上为我们工作.我们完全能够将S3中的数据读入数据帧,处理数据,从结果中创建表并使用MicroStrategy进行读取.Vinays回答也有用.

Answer 4

你需要/创建一个HiveContext

import org.apache.spark.sql.hive.HiveContext;

HiveContext sqlContext = new org.apache.spark.sql.hive.HiveContext(sc.sc());

然后直接保存数据帧或选择要存储为hive表的列

df是数据帧

df.write().mode("overwrite").saveAsTable("schemaName.tableName");

要么

df.select(df.col("col1"),df.col("col2"), df.col("col3")) .write().mode("overwrite").saveAsTable("schemaName.tableName");

要么

df.write().mode(SaveMode.Overwrite).saveAsTable("dbName.tableName");

SaveModes是Append/Ignore/Overwrite/ErrorIfExists

我在这里添加了来自Spark Documentation的HiveContext的定义,

除了基本的SQLContext之外,您还可以创建一个HiveContext,它提供了基本SQLContext提供的功能的超集.其他功能包括使用更完整的HiveQL解析器编写查询,访问Hive UDF以及从Hive表读取数据的功能.要使用HiveContext,您不需要现有的Hive设置,并且SQLContext可用的所有数据源仍然可用.HiveContext仅单独打包,以避免在默认的Spark构建中包含所有Hive的依赖项.

在Spark版本1.6.2上,使用"dbName.tableName"会出现此错误:

org.apache.spark.sql.AnalysisException:临时表不允许指定数据库名称或其他限定符.如果表名中包含点(.),请使用反引号().`引用表名

Answer 5

很抱歉写这篇文章晚了，但我看不到接受的答案。

df.write().saveAsTable会抛出AnalysisException并且与 HIVE 表不兼容。

存储 DFdf.write().format("hive")应该可以解决问题！

但是，如果这不起作用，那么按照之前的评论和答案，这是我认为最好的解决方案（尽管接受建议）。

最好的方法是显式创建 HIVE 表（包括 PARTITIONED 表），

def createHiveTable: Unit ={
spark.sql("CREATE TABLE $hive_table_name($fields) " +
  "PARTITIONED BY ($partition_column String) STORED AS $StorageType")
}

将 DF 保存为临时表，

df.createOrReplaceTempView("$tempTableName")

并插入到 PARTITIONED HIVE 表中：

spark.sql("insert into table default.$hive_table_name PARTITION($partition_column) select * from $tempTableName")
spark.sql("select * from default.$hive_table_name").show(1000,false)

当然，DF 中的LAST COLUMN将是PARTITION COLUMN，因此相应地创建 HIVE 表！

如果有效请评论！或不。

- 更新 -

df.write()
  .partitionBy("$partition_column")
  .format("hive")
  .mode(SaveMode.append)
  .saveAsTable($new_table_name_to_be_created_in_hive)  //Table should not exist OR should be a PARTITIONED table in HIVE

Answer 6

保存到Hive只是使用write()SQLContext的方法即可：

df.write.saveAsTable(tableName)

参见https://spark.apache.org/docs/2.1.0/api/java/org/apache/spark/sql/DataFrameWriter.html#saveAsTable(java.lang.String）

从Spark 2.2：使用DataSet代替DataFrame。