HDFS未处理子目录自动化批量处理方案咨询

我来帮你搞定这个需求！核心思路就是用数据库记录已处理过的HDFS子目录路径，每次运行脚本时先从数据库拉取已处理列表，对比HDFS上的所有子目录，只处理未被记录的那些。下面是具体的实现方案，包括数据库设计、脚本修改和关键细节说明：

1. 数据库表设计 #

我们先创建一个简单的表来存储已处理的HDFS子目录信息，这里以轻量易上手的SQLite为例（如果是多机器共享处理场景，可以换成MySQL/PostgreSQL）。

创建表的SQL语句：

CREATE TABLE IF NOT EXISTS processed_hdfs_dirs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    dir_path TEXT UNIQUE NOT NULL,
    processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

• dir_path设为UNIQUE，避免重复记录同一个目录；
• processed_at记录处理时间，方便后续排查处理历史。

2. 完整脚本实现 #

我们把数据库操作封装成函数，再集成到你的原有处理逻辑中。下面是修改后的完整Python脚本：

首先安装依赖（如果没装的话）：

pip install hdfs  # sqlite3一般是Python自带的，无需额外安装

然后是脚本代码：

import sh
from hdfs import InsecureClient
import sqlite3

# 数据库操作函数封装
def init_db(db_path='hdfs_processed.db'):
    """初始化数据库（第一次运行时自动创建表）"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS processed_hdfs_dirs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            dir_path TEXT UNIQUE NOT NULL,
            processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.commit()
    conn.close()

def get_processed_dirs(db_path='hdfs_processed.db'):
    """从数据库获取已处理的HDFS子目录列表"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("SELECT dir_path FROM processed_hdfs_dirs")
    processed_list = [row[0] for row in cursor.fetchall()]
    conn.close()
    return processed_list

def mark_dir_processed(dir_path, db_path='hdfs_processed.db'):
    """将处理完成的子目录标记到数据库"""
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    try:
        cursor.execute(
            "INSERT INTO processed_hdfs_dirs (dir_path) VALUES (?)",
            (dir_path,)
        )
        conn.commit()
        print(f"已标记目录 {dir_path} 为已处理")
    except sqlite3.IntegrityError:
        # 避免重复插入（比如手动重复运行脚本）
        print(f"目录 {dir_path} 已存在于处理记录中，跳过标记")
    finally:
        conn.close()

# 核心处理逻辑
def main():
    # 初始化数据库（第一次运行自动建表）
    init_db()

    # HDFS配置
    hdfs_namenode_http = 'http://VPS-DATA1:50070'  # NameNode的HTTP端口，默认50070
    hdfs_root_dir = r"hdfs://VPS-DATA1:9000/directory"

    # 初始化HDFS客户端
    hdfs_client = InsecureClient(hdfs_namenode_http)

    # 获取已处理的目录列表
    processed_dirs = get_processed_dirs()
    print(f"已处理过的目录数量: {len(processed_dirs)}")

    # 获取HDFS根目录下的所有子目录
    ls_output = sh.hdfs('dfs', '-ls', hdfs_root_dir).split('\n')
    all_subdirs = []
    for line in ls_output:
        if not line:
            continue
        # 解析-ls输出，过滤出目录（权限位第一个字符为'd'）
        if line.startswith('d'):
            parts = line.rsplit(None, 1)
            if len(parts) >= 2:
                subdir_path = parts[-1]
                all_subdirs.append(subdir_path)

    # 筛选未处理的目录
    unprocessed_dirs = [dir for dir in all_subdirs if dir not in processed_dirs]
    if not unprocessed_dirs:
        print("没有未处理的子目录，脚本结束")
        return

    print(f"发现 {len(unprocessed_dirs)} 个未处理子目录，开始处理...")

    # 批量处理未处理目录
    for dir_path in unprocessed_dirs:
        print(f"===== 开始处理目录: {dir_path} =====")
        try:
            # 替换成你的实际数据处理逻辑
            # 示例：遍历目录下的所有文件并读取
            # 注意：把HDFS路径转换为客户端可识别的相对路径（去掉hdfs://前缀）
            relative_dir = dir_path.replace('hdfs://VPS-DATA1:9000', '')
            for root, _, files in hdfs_client.walk(relative_dir):
                for file_name in files:
                    file_path = f"{root}/{file_name}"
                    with hdfs_client.open(file_path, 'r') as read_file:
                        # 这里写你的数据处理代码
                        # 比如：data = read_file.read()
                        # ...
                        print(f"已读取文件: {file_path}")

            # 处理完成后标记到数据库
            mark_dir_processed(dir_path)
        except Exception as e:
            print(f"处理目录 {dir_path} 时出错: {str(e)}")
            # 可选：如果希望出错后下次继续处理，就不要调用mark_dir_processed
            # mark_dir_processed(dir_path)

if __name__ == "__main__":
    main()

3. 关键细节说明 #

• HDFS客户端选择：示例中用了hdfs库的InsecureClient，比直接调用sh.hdfs更稳定，也更方便遍历目录；如果你的环境依赖sh.hdfs，可以保留原有的-ls解析逻辑，但要注意过滤目录（判断权限位开头为d）。
• 数据库扩展性：如果需要多机器共享处理状态，把SQLite换成MySQL/PostgreSQL即可，只需要修改init_db、get_processed_dirs、mark_dir_processed里的数据库连接逻辑。
• 错误处理：脚本中加入了异常捕获，处理出错时可以根据业务需求决定是否标记为已处理——如果希望出错后下次重试，就不要调用mark_dir_processed；如果即使出错也不想再处理，就调用该函数。
• 幂等性：数据库表的dir_path设为UNIQUE，即使重复运行脚本也不会重复记录；同时建议你的数据处理逻辑保持幂等（重复处理不会产生副作用）。

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