apache-spark - 带有自定义架构的 Spark 读取 Parquet

Question

我正在尝试使用带有自定义架构的 Parquet 格式导入数据，但它返回:
类型错误:option() 缺少 1 个必需的位置参数:“值”

   ProductCustomSchema = StructType([
        StructField("id_sku", IntegerType(), True),
        StructField("flag_piece", StringType(), True),
        StructField("flag_weight", StringType(), True),
        StructField("ds_sku", StringType(), True),
        StructField("qty_pack", FloatType(), True)])

def read_parquet_(path, schema) : 
    return spark.read.format("parquet")\
                             .option(schema)\
                             .option("timestampFormat", "yyyy/MM/dd HH:mm:ss")\
                             .load(path)

product_nomenclature = 'C:/Users/alexa/Downloads/product_nomenc'
product_nom = read_parquet_(product_nomenclature, ProductCustomSchema)

Answer 1

如评论中所述，您应该更改 .option(schema)至 .schema(schema) . option()要求您指定 key (您正在设置的选项的名称)和 value (您要为该选项分配什么值)。您正在收到 TypeError因为你只是传递了一个名为 schema 的变量至 option没有指定您实际尝试使用该变量设置的选项。
QueryExecutionException由于您在 schema 中定义的架构，您在评论中发布的信息正在被提出。变量与您的 DataFrame 中的数据不匹配。如果您要指定自定义架构，则必须确保该架构与您正在读取的数据相匹配。在您的示例中，列 id_sku存储为 BinaryType ，但在您的架构中，您将该列定义为 IntegerType . pyspark不会尝试协调您提供的架构与数据中的实际类型之间的差异，并且将引发异常。

要修复您的错误，请确保您定义的架构正确地表示您的数据，因为它存储在 Parquet 文件中(即将架构中 id_sku 的数据类型更改为 BinaryType )。这样做的好处是您不必在每次读取 Parquet 文件时都推断文件架构，从而获得轻微的性能提升。