如何将Airflow日志扫描代码改造为PySpark或多进程实现？

当然可以！针对你的Airflow日志扫描任务，不管是Python多进程还是PySpark都能帮你把任务拆分到48个处理器上，大幅压缩耗时。下面我结合你现有的代码，分别给你讲两种方案的实现思路，都是新手友好的：

一、Python多进程方案（上手更简单，适合无分布式经验的场景） #

因为你没有多进程经验，这个方案绝对是首选——不需要额外搭建集群，直接用Python自带的multiprocessing库就能把耗时的任务拆到多个CPU核心上跑，改动也很小。

核心思路 #

你的任务里最耗时的环节是遍历目录收集文件和读取文件修改时间，这两个步骤完全可以并行处理：每个进程负责处理一部分路径，最后再把结果合并起来。

改造后的代码示例 #

import os
import datetime as dt
import shlex
from subprocess import Popen, PIPE
import multiprocessing as mp
import pandas as pd

def collect_files(path):
    """单个路径下收集符合条件的.log文件"""
    files = []
    for r, d, f in os.walk(path):
        for file in f:
            # 把过滤逻辑直接整合在这里，减少后续步骤
            if '.log' in file and 'dag_processor_manager' not in file and 'scheduler' not in file:
                files.append(os.path.join(r, file))
    return files

def get_mod_time(file):
    """获取单个文件的修改时间，返回(文件路径, 修改时间)元组"""
    mod_time = dt.datetime.fromtimestamp(os.path.getmtime(file))
    return (file, mod_time)

if __name__ == '__main__':
    current_time = dt.datetime.now()
    two_hours_ago = current_time - dt.timedelta(minutes=125)
    print("Running bash command to find processes that ran within the past 2 hours...")
    
    mtime = -2/24
    bsh_cmd_find = "find /home/storage/user/airflow/logs -maxdepth 3 -type d -mtime {0} -wholename *_*/[A-Za-z]*".format(mtime)
    args = shlex.split(bsh_cmd_find)
    paths = []
    for out in Popen(args, stdout=PIPE).stdout:
        out = out.decode("utf-8").rstrip()
        paths.append(out)
    
    print("Extracting all the logs in parallel...")
    # 初始化48个进程的进程池，并行处理所有路径
    with mp.Pool(processes=48) as pool:
        # 每个进程处理一个path，收集符合条件的文件
        all_files = pool.map(collect_files, paths)
        # 把嵌套列表展平成一维列表
        all_files = [file for sublist in all_files for file in sublist]
    
    print("Extracting modification time in parallel...")
    # 再用进程池并行获取每个文件的修改时间
    with mp.Pool(processes=48) as pool:
        tuple_list = pool.map(get_mod_time, all_files)
    
    # 后续筛选逻辑和你原来的一致
    my_df = pd.DataFrame(tuple_list, columns=['Files', 'Date'])
    my_df = my_df[(my_df['Date'] >= two_hours_ago) & (my_df['Date'] <= current_time)]
    files = my_df['Files'].tolist()
    print(f"Found {len(files)} eligible log files.")

优势 #

• 零额外依赖，只用Python标准库和你已有的pandas
• 代码改动极小，只是把原来的串行循环改成了进程池调用
• 完美适配你单机器48核的场景，不需要任何集群配置

二、PySpark方案（适合大规模日志场景，需要Spark环境） #

如果你的日志量未来会持续增长，或者以后要扩展到多机器集群，PySpark会是更长远的选择。不过需要你先搭建好Spark环境（单机或集群都可以）。

核心思路 #

Spark会自动把扫描、过滤、筛选的任务分发到各个核心/节点上，你只需要用Spark的API描述任务逻辑就行，不用手动管理进程。

改造后的代码示例 #

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, udf
from pyspark.sql.types import TimestampType
import datetime as dt
import shlex
from subprocess import Popen, PIPE
import os

if __name__ == '__main__':
    current_time = dt.datetime.now()
    two_hours_ago = current_time - dt.timedelta(minutes=125)
    
    # 第一步：获取待扫描路径（和你原来的逻辑完全一致）
    print("Running bash command to find processes that ran within the past 2 hours...")
    mtime = -2/24
    bsh_cmd_find = "find /home/storage/user/airflow/logs -maxdepth 3 -type d -mtime {0} -wholename *_*/[A-Za-z]*".format(mtime)
    args = shlex.split(bsh_cmd_find)
    paths = []
    for out in Popen(args, stdout=PIPE).stdout:
        out = out.decode("utf-8").rstrip()
        paths.append(out)
    
    # 初始化SparkSession，指定用48个核心
    spark = SparkSession.builder \
        .appName("AirflowLogScanner") \
        .config("spark.executor.cores", "48") \
        .getOrCreate()
    
    # 第二步：递归扫描所有路径下的.log文件，Spark自动处理并行
    print("Extracting all the logs with Spark...")
    df = spark.read.text(paths, recursiveFileLookup=True)
    # 提取文件路径（Spark自带的元数据列）
    df = df.withColumn("file_path", col("input_file_name"))
    # 过滤掉不需要的文件
    df = df.filter(
        col("file_path").like("%.log") &
        ~col("file_path").contains("dag_processor_manager") &
        ~col("file_path").contains("scheduler")
    )
    
    # 第三步：自定义UDF获取文件修改时间
    print("Extracting modification time...")
    def get_mod_time(file_path):
        return dt.datetime.fromtimestamp(os.path.getmtime(file_path))
    mod_time_udf = udf(get_mod_time, TimestampType())
    df = df.withColumn("mod_time", mod_time_udf(col("file_path")))
    
    # 第四步：筛选近2小时内的文件
    df = df.filter(
        col("mod_time") >= two_hours_ago &
        col("mod_time") <= current_time
    )
    
    # 把结果转成Python列表
    files = df.select("file_path").rdd.flatMap(lambda x: x).collect()
    print(f"Found {len(files)} eligible log files.")
    
    spark.stop()

注意事项 #

• 需要先安装PySpark：pip install pyspark
• 如果是单机Spark，spark.executor.cores设为48即可；如果是集群，需要根据集群资源调整配置
• Spark的recursiveFileLookup=True会自动递归扫描子目录，省去了你自己写os.walk的麻烦

总结 #

• 如果你只是单机器48核的场景，优先选Python多进程方案，上手快、改动小，完全能满足你的需求，把10分钟的耗时压到几分钟以内
• 如果以后要处理海量日志或者用到多机器集群，再考虑PySpark方案

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