0

我正在尝试通过很少的转换（添加日期）将 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')
4

1 回答 1

0

我终于想出了如何用火花做到这一点。

    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，并且不会附加不同的日期。

于 2019-09-04T15:57:53.117 回答