企业官网建设流程全解析

前言

在 Python 中，subprocess 是一个非常强大的内置标准库。它的主要作用是在 Python 代码中执行外部的命令和程序（就像你在终端或命令行里敲命令一样）。

subprocess.check_output 是subprocess 模块中一个非常实用的便捷函数。它的核心作用是：执行外部命令，并直接返回该命令的标准输出（stdout）。如果命令执行失败（返回码非 0），它会自动抛出异常。

基本用法

默认情况下，check_output返回的是字节串(bytes)，在python3中，我们通常会加上text=True,让它直接返回字符串。

importsubprocess# 执行命令，获取输出# text=True 会将返回的 bytes 自动解码为 stroutput=subprocess.check_output(["echo","Hello World!"],text=True)print(output)# 输出: Hello World!

核心特性：自动检查错误(check)

函数名中的check意味着它会自动检查命令的退出状态码(returncode)。如果命令执行失败(比如命令拼错、文件不存在)，它会抛出subprocess.CalledProcessError异常。

importsubprocesstry:# 尝试查看一个不存在的文件output=subprocess.check_output(["ls","not_exist.txt"],text=True)exceptsubprocess.CalledProcessErrorase:print(f"命令执行失败！")print(f"返回码:{e.returncode}")print(f"原本的命令:{e.cmd}")

处理标准错误 (stderr)

默认情况下，check_output只返回标准输出(stdout)。如果命令产生了错误信息(stderr),这些错误信息会直接打印到屏幕上，而不会被包含在返回的变量中。
如果你相把错误信息也一并合并到输出结果中，可以使用stderr=subprocess.STDOUT

importsubprocess# 将 stderr 合并到 stdout 中一起返回output=subprocess.check_output(["ls","not_exist.txt"],stderr=subprocess.STDOUT,text=True)

实战1

importsubprocess# 1. 定义执行单条 SQL 的函数defrun_query(cmd):# 使用 check_output，如果 SQL 报错会自动抛出 CalledProcessErrortry:output=subprocess.check_output(cmd,universal_newlines=True)returnoutput.strip()exceptsubprocess.CalledProcessErrorase:print(f"Error executing Hive query:{e.output}")sys.exit(1)# 主方法defmain():#table ='test.table_test'#day ='2026-01-01'# 待查询的语句query=f"SELECT count(1) FROM{table}WHERE day ={day}"queue_name="root.queue.xa"cmd=['spark3-sql','--queue',queue_name,'-e',query]# 执行查询query_result=run_query(cmd)print(query_result)if__name__=="__main__":main()

实战2

importsubprocessimportconcurrent.futures# 1. 定义执行单条 SQL 的函数defrun_query(query,queue_name='root.queue.xa'):cmd=['spark3-sql','--queue',queue_name,'-e',query]# 使用 check_output，如果 SQL 报错会自动抛出 CalledProcessErrortry:output=subprocess.check_output(cmd,universal_newlines=True)returnoutput.strip()exceptsubprocess.CalledProcessErrorase:print(f"Error executing spark-sql query:{e.output}")sys.exit(1)# 2. 定义并发执行多条 SQL 的主函数defrun_queries_concurrent(queries_dict,max_workers=3):results={}# 使用 ThreadPoolExecutor 进行并发withconcurrent.futures.ThreadPoolExecutor(max_workers=max_workers)asexecutor:# 提交所有任务future_to_query={executor.submit(run_query,sql):keyforkey,sqlinqueries_dict.items()}# 获取结果 (as_completed 会在任务完成时立刻产出)forfutureinconcurrent.futures.as_completed(future_to_query):key=future_to_query[future]try:# 设置超时时间（例如 3600 秒），防止任务永久卡死results[key]=future.result(timeout=3600)print(f"任务 [{key}] 执行成功！")exceptExceptionase:print(f"任务 [{key}] 发生未知错误:{e}")returnresults# 主方法defmain():# 准备你的 SQL 字典my_queries={"task_1":f"SELECT count(1) FROM test.table_test where day ='2026-06-01' ","task_2":f"SELECT count(1) FROM test.table_test where day ='2026-06-02' ","task_3":f"SELECT count(1) FROM test.table_test where day ='2026-06-03' "}print("开始并发执行 Spark SQL 任务...")# 执行并发任务 (最多同时跑 3 个)final_results=run_queries_concurrent(my_queries,max_workers=3)print("-"*30)print(f"🎉 所有任务执行完毕！总耗时:{end_time-start_time:.2f}秒")task_1_count=int(final_results['task_1'].strip())task_2_count=int(final_results['task_2'].strip().split('\t')[0])task_3_count=int(final_results['task_3'].strip().split('\t')[0])# 打印结果print(f'task_1_count:{task_1_count}')print(f'task_2_count:{task_2_count}')print(f'task_3_count:{task_3_count}')if__name__=="__main__":main()