多个镶木地板文件具有 1-2 列的不同数据类型

Question

我正在尝试使用 Pyspark 将不同的 parquet 文件读入一个数据帧，它给了我错误，因为多个 parquet 文件中的某些列具有不同数据类型的列。例如： “geo”列在某些文件中具有“Double”数据类型，在其他文件中具有“String”数据类型。

我应该如何处理这个问题？我是否必须在多个 parquet 文件中手动转换具有不同数据类型的每一列？

Answer 1

背景

我在工作中遇到了一个非常相似的问题。我有一个审计文件夹，其中每个文件都有相同的列名，但 Date 列是 StringType 而不是 TimestampType。

作为背景，我最初是一名 C# 开发人员，我已经开始使用 Azure Synapse Analytics。我对使用 Pyspark 还很陌生，所以我花了几天时间解决这个问题，环顾四周，但我一直在努力寻找任何东西。我最终自己想出了一个解决方案，所以我想我会回来为你提供指导或任何处于困境中的未来开发人员。

解决方案

首先，我定义了与我的新文件和正确文件的列匹配的架构。

schema = StructType([,
    StructField("Date", TimestampType(), False),
    StructField("PipelineName", StringType(), False),
    StructField("SourceFileName", StringType(), False),
    StructField("ActivityName", StringType(), True),
    StructField("InsertedRows", LongType(), True),
    StructField("FailedRows", LongType(), True),
    StructField("Message", StringType(), True)])

然后，我使用刚刚创建的架构创建了一个空白数据框。

df = spark.createDataFrame(spark.sparkContext.emptyRDD(), schema)

然后我遍历了我的审计目录中的每个文件。

auditDir = "DirectoryNameHere"
for f in mssparkutils.fs.ls(auditDir)

在我的 for 循环中，我使用目录名和文件名来创建文件路径。

我创建了一个新的数据框，它使用查看 Date 列的 withColumn 读取此文件的路径，并将其转换为时间戳。无论该列是 StringType 还是 TimestampType，这似乎都有效。

我认为可以做类似的事情来将你的字符串转换为双精度。

因为我所有的 parquet 文件都有相同的列，所以我只使用了 union 方法而不是 unionByName，这会将这个文件的数据连接到一个不断增长的数据帧中。

path = os.path.join(auditDir, f.name)
newDf = spark.read.parquet(path).withColumn("Date", to_timestamp(col("Date")))    
df = df.union(newDf)

出于我的目的，然后我将按照最早日期的顺序将它们写成一个分区。然后将这个新分区写入同一个文件夹。但是，如果您可以写入另一个文件夹并保留旧文件作为备份，我会建议您。

df.orderBy(col("Date")).coalesce(1).write.mode("append").parquet(auditDir)

虽然迭代文件并不理想，但这是我找到的唯一方法。谢天谢地，我的数据量很小，所以没花太长时间。

感谢阅读，希望对你和其他人有很大帮助

-乔什

Answer 2

由于您尚未发布任何特定数据集。我以下面的示例向您展示在将多个镶木地板/数据帧读取到单个数据帧时事情的实际功能 -

在读取数据之前传递 Schema 的情况

from pyspark.sql.types import StructType, StructField, IntegerType, StringType

list_1 = [[1,"item 1" , 2],[2 ,"item 2", 3],[1 ,"item 3" ,2],[3 ,"item 4" , 5]]
list_2 = [[1,"item 1",2],[3 ,"item 4",2],[4 ,"item 4",4],[5 ,"item 5",2]]

my_schema = StructType([StructField("Item_ID",IntegerType(), True),StructField("Item_Name",StringType(), True ),StructField("Quantity",IntegerType(), True)])
df1 = spark.createDataFrame(list_1, my_schema)
df2 = spark.createDataFrame(list_2, my_schema)

df1.createOrReplaceTempView("df1")
df1.createOrReplaceTempView("df2")

在这里，您可以看到我在读入 df 时传递了一个模式。在您的情况下，它将来自镶木地板文件。

现在，如果我希望将上面的两个数据帧读入一个数据帧，我可以确保两个 DF 对所有相应的列都具有相同的架构。

因此，下面的代码将起作用 -

 df3 = df2.union(df1)

但在你的情况下，它没有。因此，为避免这种情况，您需要确保两个镶木地板中的每一列都应属于同一类型。因此，您可以做的是，在两个不同的数据帧中读取两个镶木地板并推断架构以进行比较。

获得类型不匹配的列信息后，您可以键入强制类型转换，然后执行上述 union() 操作以读取单个 DF 中的数据。