PySpark DataFrame 写入空（零字节）文件答案

【问题标题】：PySpark DataFrame writing empty (zero bytes) filesPySpark DataFrame 写入空（零字节）文件
【发布时间】：2022-11-02 06:06:43
【问题描述】：

我在本地设置上使用带有 Spark 版本 3.1.1 的 PySpark DataFrame API。在读入数据、执行一些转换等之后，我将DataFrame 保存到磁盘。输出目录与part-0000* 文件一起创建，并且在输出目录中也存在_SUCCESS 文件。但是，我的 part-0000* 始终为空，即零字节。

我试过用parquet 和csv 两种格式编写它，结果相同。就在写作之前，我打电话给df.show() 以确保DataFrame 中有数据。

### code.py ###
from pyspark.sql import SparkSession
from pyspark.sql import functions as F
import configs

spark = SparkSession.builder.appName('My Spark App').getOrCreate()
data = spark.read.csv(configs.dataset_path, sep=configs.data_delim)

rdd = data.rdd.map(...)
data = spark.createDataFrame(rdd)
data = data.withColumn('col1', F.lit(1))

data.show()  # Shows top 20 rows with data

data.write.parquet(save_path + '/dataset_parquet/', mode='overwrite') # Zero Bytes
data.write.csv(save_path + '/dataset_csv/', mode='overwrite')         # Zero Bytes

我正在运行此代码如下

export PYSPARK_PYTHON=python3
$SPARK_HOME/bin/spark-submit \
    --master local[*] \
    code.py

【问题讨论】：

如果您尝试将 RDD 写入磁盘会发生什么？
刚刚用rdd 测试过，结果相同。

标签： python apache-spark pyspark

【解决方案1】：

所以我遇到了与 pyspark 类似的问题，我还注意到一件事是，当我尝试将模式设置为覆盖时，它也失败了。覆盖的问题是它在写入过程中写入失败，因此它会创建文件，失败，重试，重试将失败，并显示“文件已存在”，因为它已经过了点在处理覆盖的过程中。

所以我添加了缓存来强制评估，因为就像上面的 .show() 一样，我正在执行 data.cache().count()。计数显示记录，但使用 show 或 write 进行的任何进一步评估显示 DF 为空。

因此，尝试将 .cache() 添加到该数据帧的第一个引用中，看看它是否解决了您的问题。它对我有用。

df_bad = df_cln.filter(F.col('isInvalid')).select(F.concat(F.col('line')
                       ,F.lit(">> LINE:"),F.col('monotonically_increasing_id'))
                       .alias("line"),F.col('monotonically_increasing_id'))
removed_file_cnt = df_bad.cache().count()
    print(f"The count of the records still containing udf chars in the file: {removed_file_cnt}")
if removed_file_cnt > 0:
            df_bad.coalesce(1)
            .orderBy('monotonically_increasing_id')
            .drop('monotonically_increasing_id')
            .write.option("ignoreTrailingWhiteSpace","false").option("encoding", "UTF-8")
            .format('text').save(s3_error_bucket_path, mode='overwrite')

【讨论】：