将新列附加到现有 parquet 文件答案

【问题标题】：Append a new column to an existing parquet file将新列附加到现有 parquet 文件
【发布时间】：2023-03-07 04:26:02
【问题描述】：

有没有办法在现有的 parquet 文件中添加新列？

我目前正在参加一场 kaggle 比赛，我已将所有数据转换为 parquet 文件。

情况就是这样，我将 parquet 文件读入 pyspark DataFrame，进行了一些特征提取并使用

将新列附加到 DataFrame

pysaprk.DataFrame.withColumn().

之后，我想将新列保存在源 parquet 文件中。

我知道 Spark SQL 带有 Parquet schema evolution，但该示例仅显示了带有键值的情况。

镶木地板“附加”模式也不能解决问题。它只会将新行附加到镶木地板文件中。如果有办法将新列附加到现有的镶木地板文件而不是再次生成整个表？或者我必须生成一个单独的新镶木地板文件并在运行时加入它们。

【问题讨论】：

如果您从架构上看，无法将新列附加到现有 parquet 文件中。这就像玩 parquet 文件的元数据一样。
虽然你可以尝试重写它..通过首先更改架构..我不太确定这在 spark-sql 中是如何发生的
是的，在 spark-sql 中更改模式很容易，但覆盖整个 parquet 文件的成本很高，这意味着我必须再次重新计算整个表。感谢您的评论，@AviralKumar

标签： apache-spark apache-spark-sql parquet

【解决方案1】：

在 parquet 中，您不会修改文件，而是读取、修改并写回文件，您不能只更改需要读取和写入完整文件的列。

【讨论】：

【解决方案2】：

虽然这个问题已经发布了2年，仍然没有答案，让我自己回答一下。

当我还在使用 Spark 时，Spark 的版本是 1.4。我不支持新版本，但对于那个版本，向 parquet 文件添加新列是不可能的。

【讨论】：

spark 版本 2.4.4 怎么样。我正在使用 pyspark 创建 paquet。现在我需要在现有 parquet 中添加一列。我该怎么做。

【解决方案3】：

是的，Databricks Delta 和 parquet 表都可以。下面给出一个例子：-

这个例子写在 python (pySpark)

df = sqlContext.createDataFrame([('1','Name_1','Address_1'),('2','Name_2','Address_2'),('3','Name_3','Address_3')], schema=['ID', 'Name', 'Address'])

delta_tblNm = 'testDeltaSchema.test_delta_tbl'
parquet_tblNm = 'testParquetSchema.test_parquet_tbl'

delta_write_loc = 'dbfs:///mnt/datalake/stg/delta_tblNm'
parquet_write_loc = 'dbfs:///mnt/datalake/stg/parquet_tblNm'


# DELTA TABLE
df.write.format('delta').mode('overwrite').option('overwriteSchema', 'true').save(delta_write_loc)
spark.sql(" create table if not exists {} using DELTA LOCATION '{}'".format(delta_tblNm, delta_write_loc))
spark.sql("refresh table {}".format(print(cur_tblNm)))

# PARQUET TABLE
df.write.format("parquet").mode("overwrite").save(parquet_write_loc)
spark.sql("""CREATE TABLE if not exists {} USING PARQUET LOCATION '{}'""".format(parquet_tblNm, parquet_write_loc))
spark.sql(""" REFRESH TABLE {} """.format(parquet_tblNm))

test_df = spark.sql("select * testDeltaSchema.test_delta_tbl")
test_df.show()

test_df = spark.sql("select * from testParquetSchema.test_parquet_tbl")
test_df.show()

test_df = spark.sql("ALTER TABLE  testDeltaSchema.test_delta_tbl ADD COLUMNS (Mob_number String COMMENT 'newCol' AFTER Address)")
test_df.show()

test_df = spark.sql("ALTER TABLE  testParquetSchema.test_parquet_tbl ADD COLUMNS (Mob_number String COMMENT 'newCol' AFTER Address)")
test_df.show()

【讨论】：