sql - 以 parquet 格式保存数据框会生成太多小文件

Question

使用 Spark Sql，我将 parquet 格式的 spark df 保存在配置单元表中。问题是此操作总是生成大约 200 个 3 MB 的 part-* 文件。在处理过程中如何控制输出文件的数量和分区的数量？据我所知，每个分区生成一个 part-* 文件。

我的工作流程是怎样的:

1. 我使用 spark.read.parquet("")
从 Hive 元存储中的表中读取一个分区
2. 然后我从这个 df 创建一个 TempView
3. 然后我用纯 sql 进行我需要的转换，它有很多连接:spark.sql("SQL QUERY")
4. 然后我尝试对结果进行重新分区 或合并，但这不起作用，因为 part-* 文件的数量始终为 200
5. 我用 spark.sql("insert overwrite partition() ... ")
保存我的 df

我使用纯 sql 的原因是，如果尝试使用 Dataframe API 保存我的 df，我会丢失列的顺序。

我在 CDH 11.2 中使用 Spark 2.2.0

如果我的问题中有一些非慈善机构，请告诉我。

UPDATE1:示例源代码

conf = (SparkConf()
.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
.set("spark.sql.parquet.compression.codec","SNAPPY")
.set("spark.executor.cores", "3")
.set("spark.executor.instances", "10")
.set("spark.executor.memory", "20g")
)

sc = SparkContext(conf=conf)

spark = SparkSession \
    .builder \
    .config(conf=conf) \
    .getOrCreate()

df=spark.read.parquet("/user/hive/warehouse/schema.db/tbl/year=2000/month=01/day=01").cache()

temp_table_name="tmp_bal2_nor_20000101"
df.createOrReplaceTempView(temp_table_name)

aaa=spark.sql(
""" select DISTINCT t.A,
                    t.lot,
                    t.of,
                    t.columns,
                    .
                    .,
    from temp_table_name t 
left join( 
    ...
) tmp1 on ...
left join( 
    ...
) tmp2 on ...
left join( 
    ...
) tmp3 ...
left join( 
    ...
) tmp4 on ...
left join( 
    ...
) tmp5 on ...
left join( 
    ...
) tmp6 on ...
order by t.A, t.lot """
)

aaa.coalesce(10).createOrReplaceTempView(temp_table_name)
spark.sql(""" insert overwrite table tbl1
PARTITION(year=2000, month=01, day=01)
select DISTINCT t.A, 
                 t.lot, 
                 t.of, 
                 t.columns
from temp_table_name t
""")

spark.stop()

Answer 1

尝试更改您的代码，以便在使用 coalesce 后保存 parquet 文件。如下:

 aaa.coalesce(10).write.parquet("your_path")

否则 coalesce 通常会被忽略，因为在减少分区后，您正在使用 aaa 数据集创建 TempView。目前您有 200 个分区，因为 spark.sql.shuffle.partitions 设置为 200，这是 Spark SQL 的默认分区数。 coalesce 只是您数据集的一部分，因此如果您想正确使用它，您需要直接通过您的数据集进行操作。