如何在向HDFS/S3写入分区数据集时为每个分区目录生成_SUCCESS文件？

当然有办法实现每个子分区目录生成对应的_SUCCESS文件！下面根据不同的使用场景，分享几种实用的方案：

1. Spark自定义逻辑（最常用场景） #

如果你用Spark写入分区数据，默认只会在根目录生成_SUCCESS，但可以通过Hadoop的FileSystem API在数据写入完成后，遍历所有分区目录并创建对应文件。

举个Python版本的实现示例：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("AddPartitionSuccessFiles").getOrCreate()
sc = spark.sparkContext

# 替换为你的实际输出路径（HDFS/S3都支持）
output_path = "s3://your-bucket/your-dataset"  # 或 hdfs://namenode:8020/your-dataset

# 第一步：写入分区数据（这里以按date分区为例）
df.write.partitionBy("date").parquet(output_path, mode="overwrite")

# 第二步：获取Hadoop文件系统实例
fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(sc._jsc.hadoopConfiguration())
root_path = sc._jvm.org.apache.hadoop.fs.Path(output_path)

# 遍历所有子分区目录，创建_SUCCESS
for status in fs.listStatus(root_path):
    dir_path = status.getPath()
    # 跳过根目录的系统文件（比如已存在的_SUCCESS），只处理分区子目录
    if status.isDirectory() and not dir_path.getName().startswith("_"):
        success_file = dir_path.toString() + "/_SUCCESS"
        success_path = sc._jvm.org.apache.hadoop.fs.Path(success_file)
        if not fs.exists(success_path):
            fs.createNewFile(success_path)
            print(f"Created _SUCCESS in {dir_path.toString()}")

说明：这段逻辑要在Spark Driver端执行，确保数据完全写入后再触发，避免出现分区数据未写完但_SUCCESS已存在的情况。

2. Hadoop命令行/Shell脚本方案 #

如果是通过Hadoop工具（比如MapReduce、DistCp）写入数据，可以用Shell脚本批量创建分区目录的_SUCCESS：

# 替换为你的输出路径
OUTPUT_PATH="hdfs://namenode:8020/your-dataset"

# 先执行数据写入操作（比如hadoop jar或spark-submit）

# 遍历所有分区子目录并创建_SUCCESS
hadoop fs -ls $OUTPUT_PATH | grep "^d" | awk '{print $8}' | while read dir; do
    SUCCESS_FILE="$dir/_SUCCESS"
    if ! hadoop fs -test -e $SUCCESS_FILE; then
        hadoop fs -touchz $SUCCESS_FILE
        echo "Created _SUCCESS in $dir"
    fi
done

3. S3专属方案（AWS CLI/SDK） #

针对S3对象存储，因为没有真正的目录概念，我们可以通过前缀识别分区，用AWS工具创建_SUCCESS对象：

用AWS CLI实现 #

BUCKET="your-bucket"
PREFIX="your-dataset/"

# 先完成数据写入操作

# 遍历所有分区前缀并创建_SUCCESS
aws s3 ls s3://$BUCKET/$PREFIX --recursive | grep -E "date=[0-9-]+/$" | awk '{print $4}' | while read prefix; do
    SUCCESS_KEY="$prefix_SUCCESS"
    if ! aws s3 ls s3://$BUCKET/$SUCCESS_KEY > /dev/null 2>&1; then
        # 创建空的_SUCCESS对象
        aws s3 cp /dev/null s3://$BUCKET/$SUCCESS_KEY
        echo "Created _SUCCESS in s3://$BUCKET/$prefix"
    fi
done

用Python boto3 SDK实现 #

import boto3

s3 = boto3.client('s3')
bucket = 'your-bucket'
prefix = 'your-dataset/'

# 列出所有分区前缀
response = s3.list_objects_v2(Bucket=bucket, Prefix=prefix, Delimiter='/')
for common_prefix in response.get('CommonPrefixes', []):
    partition_prefix = common_prefix['Prefix']
    success_key = f"{partition_prefix}_SUCCESS"
    # 检查文件是否存在，不存在则创建
    try:
        s3.head_object(Bucket=bucket, Key=success_key)
        print(f"_SUCCESS already exists in {partition_prefix}")
    except s3.exceptions.ClientError as e:
        if e.response['Error']['Code'] == '404':
            s3.put_object(Bucket=bucket, Key=success_key, Body=b'')
            print(f"Created _SUCCESS in {partition_prefix}")

• 关键注意事项：
  • 务必在数据写入完全成功后执行创建_SUCCESS的逻辑，避免数据不完整但标记为成功的情况。
  • 分布式场景下（比如Spark），尽量在Driver端执行文件创建，避免多Executor重复操作。
  • S3场景下，要根据你的分区格式调整前缀过滤规则（比如例子中的date=[0-9-]+/）。

内容的提问来源于stack exchange，提问作者femibyte