Pyspark：在python中将所有压缩的csv合并为一个csv答案

【问题标题】：Pyspark: Merge all zipped csvs into one csv in pythonPyspark：在python中将所有压缩的csv合并为一个csv
【发布时间】：2019-07-16 00:18:30
【问题描述】：

如果我有压缩 csv 格式的海量数据，我如何将其组合成单个 csv 文件（压缩输出与否无关紧要）？

我正在将它读入 spark Dataframes，但后来我被困在如何连接 pyspark Dataframes 上。

下面是我的代码，它运行一个循环并希望为每个循环运行附加 Dataframe：

        schema=StructType([])
        result = spark.createDataFrame(sc.emptyRDD(), schema)
        for day in range(1,31):
            day_str = str(day) if day>=10 else "0"+str(day)
            print 'Ingesting %s' % day_str
            df = spark.read.format("csv").option("header", "false").option("delimiter", "|").option("inferSchema", "true").load("s3a://key/201811%s" % (day_str))
            result = result.unionAll(df)

        result.write.save("s3a://key/my_result.csv", format='csv')

这给了我错误AnalysisException: u"Union can only be performed on tables with the same number of columns, but the first table has 0 columns and the second table has 1 columns;;\n'Union\n:- LogicalRDD\n+- Relation[_c0#75] csv\n"。谁能帮助我如何继续？

【问题讨论】：

我对 pyspark 一无所知，但是您提到的错误消息向我表明，您必须首先使每个 csv 符合要求，以便将具有相同形状的 Dataframes 组合在一起。也许您需要阅读每一个，然后将其更改为可以获取所有 Dataframe 的通用形状，以便将它们组合起来。
我不会对 S3 上的数据使用 .option("inferSchema")。 Spark 将读取整个文件以推断该模式。它不仅使您的执行时间加倍，而且使您的成本加倍。执行一次，记录推断的模式，将其放入代码中以备下次使用类似：最好保存为 Parquet 或 ORC 等格式以进行后续查询

标签： python pyspark

【解决方案1】：

这对我有用：

result=spark.createDataFrame(sc.emptyRDD(), schema_mw)

for day in range(1,31):
    day_str = str(day) if day>=10 else "0"+str(day)
    print 'Ingesting %s' % day_str

    df = spark.read.format("csv").option("header", "false").option("delimiter", ",").schema(schema_mw).load("s3a://bucket/201811%s" % (day_str))

    if result:
        result = result.union(df)
    else:
        result = df
result.repartition(1).write.save("s3a://bucket/key-Compiled", format='csv', header=False)

但是，当我尝试在重新分区的最后一步中将标头加载为 true 时，标头会存储为一行。我不确定如何将这些标题添加为标题而不是一行。

【讨论】：