前言在 Python 中subprocess 是一个非常强大的内置标准库。它的主要作用是在 Python 代码中执行外部的命令和程序就像你在终端或命令行里敲命令一样。subprocess.check_output 是subprocess 模块中一个非常实用的便捷函数。它的核心作用是执行外部命令并直接返回该命令的标准输出stdout。如果命令执行失败返回码非 0它会自动抛出异常。基本用法默认情况下check_output返回的是字节串(bytes)在python3中我们通常会加上textTrue,让它直接返回字符串。importsubprocess# 执行命令获取输出# textTrue 会将返回的 bytes 自动解码为 stroutputsubprocess.check_output([echo,Hello World!],textTrue)print(output)# 输出: Hello World!核心特性自动检查错误(check)函数名中的check意味着它会自动检查命令的退出状态码(returncode)。如果命令执行失败(比如命令拼错、文件不存在)它会抛出subprocess.CalledProcessError异常。importsubprocesstry:# 尝试查看一个不存在的文件outputsubprocess.check_output([ls,not_exist.txt],textTrue)exceptsubprocess.CalledProcessErrorase:print(f命令执行失败)print(f返回码:{e.returncode})print(f原本的命令:{e.cmd})处理标准错误 (stderr)默认情况下check_output只返回标准输出(stdout)。如果命令产生了错误信息(stderr),这些错误信息会直接打印到屏幕上而不会被包含在返回的变量中。如果你相把错误信息也一并合并到输出结果中可以使用stderrsubprocess.STDOUTimportsubprocess# 将 stderr 合并到 stdout 中一起返回outputsubprocess.check_output([ls,not_exist.txt],stderrsubprocess.STDOUT,textTrue)实战1importsubprocess# 1. 定义执行单条 SQL 的函数defrun_query(cmd):# 使用 check_output如果 SQL 报错会自动抛出 CalledProcessErrortry:outputsubprocess.check_output(cmd,universal_newlinesTrue)returnoutput.strip()exceptsubprocess.CalledProcessErrorase:print(fError executing Hive query:{e.output})sys.exit(1)# 主方法defmain():#table test.table_test#day 2026-01-01# 待查询的语句queryfSELECT count(1) FROM{table}WHERE day {day}queue_nameroot.queue.xacmd[spark3-sql,--queue,queue_name,-e,query]# 执行查询query_resultrun_query(cmd)print(query_result)if__name____main__:main()实战2importsubprocessimportconcurrent.futures# 1. 定义执行单条 SQL 的函数defrun_query(query,queue_nameroot.queue.xa):cmd[spark3-sql,--queue,queue_name,-e,query]# 使用 check_output如果 SQL 报错会自动抛出 CalledProcessErrortry:outputsubprocess.check_output(cmd,universal_newlinesTrue)returnoutput.strip()exceptsubprocess.CalledProcessErrorase:print(fError executing spark-sql query:{e.output})sys.exit(1)# 2. 定义并发执行多条 SQL 的主函数defrun_queries_concurrent(queries_dict,max_workers3):results{}# 使用 ThreadPoolExecutor 进行并发withconcurrent.futures.ThreadPoolExecutor(max_workersmax_workers)asexecutor:# 提交所有任务future_to_query{executor.submit(run_query,sql):keyforkey,sqlinqueries_dict.items()}# 获取结果 (as_completed 会在任务完成时立刻产出)forfutureinconcurrent.futures.as_completed(future_to_query):keyfuture_to_query[future]try:# 设置超时时间例如 3600 秒防止任务永久卡死results[key]future.result(timeout3600)print(f任务 [{key}] 执行成功)exceptExceptionase:print(f任务 [{key}] 发生未知错误:{e})returnresults# 主方法defmain():# 准备你的 SQL 字典my_queries{task_1:fSELECT count(1) FROM test.table_test where day 2026-06-01 ,task_2:fSELECT count(1) FROM test.table_test where day 2026-06-02 ,task_3:fSELECT count(1) FROM test.table_test where day 2026-06-03 }print(开始并发执行 Spark SQL 任务...)# 执行并发任务 (最多同时跑 3 个)final_resultsrun_queries_concurrent(my_queries,max_workers3)print(-*30)print(f 所有任务执行完毕总耗时:{end_time-start_time:.2f}秒)task_1_countint(final_results[task_1].strip())task_2_countint(final_results[task_2].strip().split(\t)[0])task_3_countint(final_results[task_3].strip().split(\t)[0])# 打印结果print(ftask_1_count:{task_1_count})print(ftask_2_count:{task_2_count})print(ftask_3_count:{task_3_count})if__name____main__:main()
