Spark如何并行执行HDFS上的Hive查询并生成含统计列的DataFrame

实现思路 #

原有代码的核心瓶颈是driver侧串行提交任务，每次必须等前一个查询跑完才会提交下一个，2400个任务串行执行耗时必然很长。我们可以在driver侧用多线程池并发提交Spark查询任务，利用集群资源并行运行多个查询，大幅压缩总耗时。

注意事项 #

• 不要尝试在executor侧（比如rdd.map/自定义UDF内部）执行spark.sql，executor侧无法直接访问driver端的SparkSession实例，运行时会直接报错
• 并发度不要设置过高，避免给集群调度带来过大压力，建议根据集群空闲资源设置为20~100之间

具体实现代码 #

from concurrent.futures import ThreadPoolExecutor, as_completed
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, LongType

# 初始化SparkSession
spark = SparkSession.builder.appName("ParallelQueryCount").getOrCreate()

# 读取输入CSV
input_df = spark.read.options(header='True', delimiter='|', inferSchema='True').csv("/input/vikas/input.csv")
input_list = input_df.collect()

# 定义单查询处理逻辑
def process_single_query(row):
    name = row.name
    script_path = row.script
    # 读取HDFS上的SQL文件内容，多行SQL拼接为完整字符串
    query_rows = spark.read.text(script_path).collect()
    query_content = '\n'.join([r[0] for r in query_rows])
    # 执行SQL统计行数
    df = spark.sql(query_content)
    count = df.count()
    # 触发告警逻辑
    if count > 0:
        call_alert_function(name, count)
    return (name, script_path, count)

# 配置并发度，可根据集群资源调整
MAX_WORKERS = 50
result_list = []

# 提交并行任务
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
    future_to_row = {executor.submit(process_single_query, row): row for row in input_list}
    for future in as_completed(future_to_row):
        try:
            res = future.result()
            result_list.append(res)
        except Exception as e:
            # 异常处理：记录失败任务，可按业务需求补默认值
            row = future_to_row[future]
            print(f"查询{row.name}执行失败，路径{row.script}，错误信息：{str(e)}")
            result_list.append((row.name, row.script, -1))

# 构造最终输出的DataFrame
output_schema = StructType([
    StructField("name", StringType(), nullable=False),
    StructField("script", StringType(), nullable=False),
    StructField("count", LongType(), nullable=False)
])
output_df = spark.createDataFrame(result_list, schema=output_schema)

# 可根据需要将结果写出到目标存储
# output_df.write.options(header='True', delimiter='|').csv("/path/to/result/output")

原有代码Bug修正说明 #

你之前实现中读取SQL文件的逻辑存在问题：spark.read.options(header='False', delimiter='|').csv(row.script) 是把SQL文件当做CSV格式解析，无法拿到完整的SQL文本内容，需要改为用spark.read.text读取SQL文本再拼接执行。

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