带有 hive 的 pyspark - 无法正确创建分区并从数据框中保存表

Posted 2023-03-31

【中文标题】带有 hive 的 pyspark - 无法正确创建分区并从数据框中保存表

【英文标题】：pyspark with hive - can't properly create with partition and save a table from a dataframe

【发布时间】：2019-09-04 13:31:03

【问题描述】：

我正在尝试通过很少的转换（添加日期）将 json 文件转换为镶木地板，但我需要先对这些数据进行分区，然后再将其保存到镶木地板。

我在这个区域碰壁了。

下面是表的创建过程：

    df_temp = spark.read.json(data_location) \
        .filter(
            cond3
        )
    df_temp = df_temp.withColumn("date", fn.to_date(fn.lit(today.strftime("%Y-%m-%d"))))
    df_temp.createOrReplaceTempView("_tmp".format("duration_small"))

    spark.sql("CREATE TABLE IF NOT EXISTS 1 LIKE 0_tmp LOCATION '2/1'".format("duration_small","duration", warehouse_location))
    spark.sql("DESC ".format("duration"))

那么关于转换的保存：

    df_final.write.mode("append").format("parquet").partitionBy("customer_id", "date").saveAsTable('duration')

但这会产生以下错误：

pyspark.sql.utils.AnalysisException: '\n指定的分区与现有表 default.duration 的不匹配。\n指定的分区列：[customer_id, date]\n现有的分区列：[]\n ;'

架构是：

    root
     |-- action_id: string (nullable = true)
     |-- customer_id: string (nullable = true)
     |-- duration: long (nullable = true)
     |-- initial_value: string (nullable = true)
     |-- item_class: string (nullable = true)
     |-- set_value: string (nullable = true)
     |-- start_time: string (nullable = true)
     |-- stop_time: string (nullable = true)
     |-- undo_event: string (nullable = true)
     |-- year: integer (nullable = true)
     |-- month: integer (nullable = true)
     |-- day: integer (nullable = true)
     |-- date: date (nullable = true)

因此我尝试将创建表更改为：

    spark.sql("CREATE TABLE IF NOT EXISTS 1 LIKE 0_tmp PARTITIONED BY (customer_id, date) LOCATION '2/1'".format("duration_small","duration", warehouse_location))

但这会产生如下错误：

...不匹配的输入“PARTITIONED”期待...

所以我发现 PARTITIONED BY 不适用于LIKE，但我的想法已经不多了。 如果使用 USING 而不是 LIKE 我得到了错误：

pyspark.sql.utils.AnalysisException: '当未定义表架构时，不允许指定分区列。当未提供表架构时，将推断架构和分区列。;'

建表时如何添加分区？

Ps - 一旦使用分区定义了表的架构，我想简单地使用：

    df_final.write.format("parquet").insertInto('duration')

【问题讨论】：

是否已经定义了工期表？那么它没有分区，但您尝试使用分区附加数据。

嗯，它是在 CREATE TABLE 上定义的，我正在尝试弄清楚如何使用分区创建它

【参考方案1】：

我终于想通了如何用 spark 做到这一点。

    df_temp.read.json...

    df_temp.createOrReplaceTempView("_tmp".format("duration_small"))

    spark.sql("""
    CREATE TABLE IF NOT EXISTS 1
    USING PARQUET
    PARTITIONED BY (customer_id, date)
    LOCATION '2/1' AS SELECT * FROM 0_tmp
    """.format("duration_small","duration", warehouse_location))

    spark.sql("DESC ".format("duration"))

    df_temp.write.mode("append").partitionBy("customer_id", "date").saveAsTable('duration')

我不知道为什么，但如果我不能使用 insertInto，它会突然使用一个奇怪的 customer_id，并且不会附加不同的日期。