如何使用Azure Data Factory检查数据湖目录下的文件是否为空

方法一：Python 实现（跨平台，支持复杂目录结构） #

如果你的数据湖是本地文件系统或可挂载的分布式文件系统（如HDFS挂载目录），可以用Python快速实现需求：

import os
import csv

# 替换为你的数据湖根目录路径
DATA_LAKE_DIR = "/path/to/your/datalake"
# 输出的CSV文件路径
OUTPUT_CSV = "empty_files.csv"

empty_files = []

# 递归遍历目录
for root, dirs, files in os.walk(DATA_LAKE_DIR):
    for file in files:
        file_path = os.path.join(root, file)
        # 判断文件是否为空（大小为0字节）
        if os.path.getsize(file_path) == 0:
            # 存储完整文件路径，也可改为仅文件名
            empty_files.append({"file_path": file_path})

# 写入CSV文件
with open(OUTPUT_CSV, mode='w', newline='', encoding='utf-8') as csv_file:
    fieldnames = ['file_path']
    writer = csv.DictWriter(csv_file, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerows(empty_files)

print(f"检查完成，共找到{len(empty_files)}个空文件，结果已保存至{OUTPUT_CSV}")

适配对象存储数据湖（如S3、OSS） #

如果数据湖是云对象存储服务，需改用对应SDK处理，以AWS S3为例：

import boto3
import csv

s3 = boto3.client('s3')
bucket_name = "your-bucket-name"
OUTPUT_CSV = "empty_s3_files.csv"
empty_files = []

# 分页遍历桶内所有对象
paginator = s3.get_paginator('list_objects_v2')
for page in paginator.paginate(Bucket=bucket_name):
    for obj in page['Contents']:
        if obj['Size'] == 0:
            empty_files.append({"s3_path": f"s3://{bucket_name}/{obj['Key']}"})

# 写入CSV
with open(OUTPUT_CSV, mode='w', newline='', encoding='utf-8') as csv_file:
    writer = csv.DictWriter(csv_file, fieldnames=['s3_path'])
    writer.writeheader()
    writer.writerows(empty_files)

方法二：Shell 脚本实现（Linux/Unix环境快速处理） #

针对Linux/Unix环境下的本地或挂载数据湖，用Shell脚本更高效：

#!/bin/bash

# 替换为数据湖目录路径
DATA_LAKE_DIR="/path/to/your/datalake"
# 输出CSV路径
OUTPUT_CSV="empty_files.csv"

# 写入CSV表头
echo "file_path" > $OUTPUT_CSV

# 递归查找空文件并追加到CSV
find $DATA_LAKE_DIR -type f -empty -print >> $OUTPUT_CSV

echo "检查完成，结果已保存至$OUTPUT_CSV"

参数说明： #

• -type f：仅查找文件（排除目录）
• -empty：筛选空文件
• 若只需文件名而非完整路径，可将-print替换为-printf "%f\n"

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