如何使用 Spark 数据框架的架构创建 Hive 表?

问题描述 投票:0回答:6

我想使用 Spark 数据帧的架构创建一个 Hive 表。我怎样才能做到这一点?

对于固定列,我可以使用:

val CreateTable_query = "Create Table my table(a string, b string, c double)"
sparksession.sql(CreateTable_query) 

但是我的数据框中有很多列,那么有没有办法自动生成这样的查询?

scala apache-spark hive
6个回答
28
投票

假设您使用的是 Spark 2.1.0 或更高版本,并且 my_DF 是您的数据框,

//get the schema split as string with comma-separated field-datatype pairs
StructType my_schema = my_DF.schema();
String columns = Arrays.stream(my_schema.fields())
                       .map(field -> field.name()+" "+field.dataType().typeName())
                       .collect(Collectors.joining(","));

//drop the table if already created
spark.sql("drop table if exists my_table");
//create the table using the dataframe schema
spark.sql("create table my_table(" + columns + ") 
    row format delimited fields terminated by '|' location '/my/hdfs/location'");
    //write the dataframe data to the hdfs location for the created Hive table
    my_DF.write()
    .format("com.databricks.spark.csv")
    .option("delimiter","|")
    .mode("overwrite")
    .save("/my/hdfs/location");

使用临时表的另一种方法

my_DF.createOrReplaceTempView("my_temp_table");
spark.sql("drop table if exists my_table");
spark.sql("create table my_table as select * from my_temp_table");

10
投票

根据您的问题,您似乎想使用数据框的架构在配置单元中创建表。但正如您所说,该数据框中有很多列,因此有两个选项

  • 第一个是通过数据框创建直接配置单元表。
  • 第二步是采用该数据帧的模式并在配置单元中创建表。

考虑这段代码:

package hive.example

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.Row
import org.apache.spark.sql.SparkSession

object checkDFSchema extends App {
  val cc = new SparkConf;
  val sc = new SparkContext(cc)
  val sparkSession = SparkSession.builder().enableHiveSupport().getOrCreate()
  //First option for creating hive table through dataframe 
  val DF = sparkSession.sql("select * from salary")
  DF.createOrReplaceTempView("tempTable")
  sparkSession.sql("Create table yourtable as select * form tempTable")
  //Second option for creating hive table from schema
  val oldDFF = sparkSession.sql("select * from salary")
  //Generate the schema out of dataframe  
  val schema = oldDFF.schema
  //Generate RDD of you data 
  val rowRDD = sc.parallelize(Seq(Row(100, "a", 123)))
  //Creating new DF from data and schema 
  val newDFwithSchema = sparkSession.createDataFrame(rowRDD, schema)
  newDFwithSchema.createOrReplaceTempView("tempTable")
  sparkSession.sql("create table FinalTable AS select * from tempTable")
}

9
投票

另一种方法是使用 StructType..sql、simpleString、TreeString 等上可用的方法...

您可以从 Dataframe 的架构创建 DDL,可以从 DDL 创建 Dataframe 的架构..

这是一个示例 -(直到 Spark 2.3)

    // Setup Sample Test Table to create Dataframe from
    spark.sql(""" drop database hive_test cascade""")
    spark.sql(""" create database hive_test""")
    spark.sql("use hive_test")
    spark.sql("""CREATE TABLE hive_test.department(
    department_id int ,
    department_name string
    )    
    """)
    spark.sql("""
    INSERT INTO hive_test.department values ("101","Oncology")    
    """)

    spark.sql("SELECT * FROM hive_test.department").show()

/***************************************************************/

现在我可以使用 Dataframe 了。在实际情况下,您将使用数据帧读取器从文件/数据库创建数据帧。让我们使用它的模式来创建 DDL

  // Create DDL from Spark Dataframe Schema using simpleString function

 // Regex to remove unwanted characters    
    val sqlrgx = """(struct<)|(>)|(:)""".r
 // Create DDL sql string and remove unwanted characters

    val sqlString = sqlrgx.replaceAllIn(spark.table("hive_test.department").schema.simpleString, " ")

// Create Table with sqlString
   spark.sql(s"create table hive_test.department2( $sqlString )")

Spark 2.4 开始,您可以在 StructType 上使用 fromDDL 和 toDDL 方法 -

val fddl = """
      department_id int ,
      department_name string,
      business_unit string
      """


    // Easily create StructType from DDL String using fromDDL
    val schema3: StructType = org.apache.spark.sql.types.StructType.fromDDL(fddl)


    // Create DDL String from StructType using toDDL
    val tddl = schema3.toDDL

    spark.sql(s"drop table if exists hive_test.department2 purge")

   // Create Table using string tddl
    spark.sql(s"""create table hive_test.department2 ( $tddl )""")

    // Test by inserting sample rows and selecting
    spark.sql("""
    INSERT INTO hive_test.department2 values ("101","Oncology","MDACC Texas")    
    """)
    spark.table("hive_test.department2").show()
    spark.sql(s"drop table hive_test.department2")


5
投票

从spark 2.4开始,您可以使用该函数来获取列名称和类型(即使对于嵌套结构)

val df = spark.read....

df.schema.toDDL

4
投票

这是从 parquet 文件创建 Hive 表的 PySpark 版本。您可能已经使用推断的架构生成了 Parquet 文件,现在想要将定义推送到 Hive 元存储。您还可以将定义推送到 AWS Glue 或 AWS Athena 等系统,而不仅仅是 Hive 元存储。这里我使用spark.sql来推送/创建永久表。

 # Location where my parquet files are present.
 df = spark.read.parquet("s3://my-location/data/")

    cols = df.dtypes
    buf = []
    buf.append('CREATE EXTERNAL TABLE test123 (')
    keyanddatatypes =  df.dtypes
    sizeof = len(df.dtypes)
    print ("size----------",sizeof)
    count=1;
    for eachvalue in keyanddatatypes:
        print count,sizeof,eachvalue
        if count == sizeof:
            total = str(eachvalue[0])+str(' ')+str(eachvalue[1])
        else:
            total = str(eachvalue[0]) + str(' ') + str(eachvalue[1]) + str(',')
        buf.append(total)
        count = count + 1

    buf.append(' )')
    buf.append(' STORED as parquet ')
    buf.append("LOCATION")
    buf.append("'")
    buf.append('s3://my-location/data/')
    buf.append("'")
    buf.append("'")
    ##partition by pt
    tabledef = ''.join(buf)

    print "---------print definition ---------"
    print tabledef
    ## create a table using spark.sql. Assuming you are using spark 2.1+
    spark.sql(tabledef);

0
投票

使用 Spark 数据帧在 Spark shell 中尝试以下方法。

/* 从任意文件读取数据*/

val df=spark.read.parquet("/test/sample/data/")

/* 检查架构 */

val schema1 = df.schema

/* 获取列和数据类型 */

val columns = schema1.fields.map(field => s"${field.name} ${field.dataType.typeName}").mkString(",")

/* 创建 hive 表 */

spark.sql("CREATE TABLE sample_table($columns)
stored as parquet
location '/test/sample/parsed/data'")
最新问题
© www.soinside.com 2019 - 2024. All rights reserved.