python - 导入pyspark ETL模块并使用python子进程作为子进程运行时出错

Question

我正在尝试使用导入模块和子进程从一个 main.py python 脚本动态调用 pyspark 模块列表。我试图调用的子模块不返回任何内容，它只是执行其 ETL 操作。我希望我的 main.py 程序等到子进程完成。在下面的代码中，每次我尝试调用子进程时，都会遇到错误“TypeError: 'NoneType' object is not iterable”。另一个问题是，在启动 subprocess.Popen 之后，我认为流程将继续在 main.py 中继续到下一行，直到它遇到 j1.wait()，但立即打印语句 (print("etl_01_job is running ") 没有执行，我错过了什么吗？
我用谷歌搜索并尝试了很多其他方法，但没有任何效果。任何人都可以阐明我做错了什么吗？一旦我能够成功调用子进程，我必须根据子进程的返回码添加一些其他条件。但在这一点上，我想解决这个问题。谢谢
主文件

import json
import importlib
import subprocess
from datetime import datetime
from pyspark import SparkContext, SparkConf
from pyspark.sql.session import SparkSession


def main():
    with open('C:/Pyspark/test/config/config.json', 'r') as config_file:
        config = json.load(config_file)

    spark = SparkSession.builder\
        .appName(config.get("app_name"))\
        .getOrCreate()

    job_module1 = importlib.import_module("etl_01_job")
    print("main calling time :", datetime.now())
    j1 = subprocess.Popen(job_module1.run_etl_01_job(spark, config))
    print("etl_01_job is running")
    j1.wait() #I'm expecting the main.py to wait until child process finishes
    print("etl_01_job finished")

    job_module2 = importlib.import_module("etl_02_job")
    j2 = subprocess.Popen(job_module2.run_etl_02_job(spark, config))

if __name__ == "__main__":
    main()

Child pyspark job:etl_01_job.py :不是原始代码，只是一个示例脚本

from datetime import datetime
import time
import sys

def etl_01_job(spark, config):
    print("I'm in 01etljob")
    print(config)
    print(config.get("app_name"))
    time.sleep(10)
    print("etljob 1 ending time :", datetime.now())
def run_etl_01_job(spark, config):
    etl_01_job(spark, config)

我得到的错误是

Traceback (most recent call last):
  File "C:/py_spark/src/main.py", line 49, in <module>
    main()
  File "C:/py_spark/src/main.py", line 38, in main
    p1 = subprocess.run(job_module1.run_etl_01_job(spark, config))
  File "C:\ProgramData\Anaconda3\lib\subprocess.py", line 489, in run
    with Popen(*popenargs, **kwargs) as process:
  File "C:\ProgramData\Anaconda3\lib\subprocess.py", line 854, in __init__
    self._execute_child(args, executable, preexec_fn, close_fds,
  File "C:\ProgramData\Anaconda3\lib\subprocess.py", line 1247, in _execute_child
    args = list2cmdline(args)
  File "C:\ProgramData\Anaconda3\lib\subprocess.py", line 549, in list2cmdline
    for arg in map(os.fsdecode, seq):
TypeError: 'NoneType' object is not iterable

Answer 1

原因是subprocess.Popen(job_module1.run_etl_01_job(spark, config))不是创建正在运行的子进程的方法 job_module1.run_etl_01_job带参数 (spark, config)) .你在这里所做的只是运行 job_module1.run_etl_01_job(spark, config)本地(不在不同的进程中)返回 None ，然后拨打 subprocess.Popen(None)这反过来又会给你你发布的错误。
您可能打算做的是使用 multiprocessing python模块如

from multiprocessing import Process

p1 = Process(target=job_module1.run_etl_01_job, args=(spark, config))
p2 = Process(target=job_module2.run_etl_02_job, args=(spark, config))
p1.start()
p2.start()
p1.join()
p2.join()

那就是说我认为它不会起作用，因为对 multiprocessing 的争论被酸洗，我认为 SparkSession 对象不会在酸洗/解压后幸存下来。您可能还想尝试使用 threading模块在这里代替甚至更方便 concurrent.futures像这样的模块:

from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor() as e:
  future1 = e.submit(job_module1.run_etl_01_job, spark, config)
  future2 = e.submit(job_module2.run_etl_02_job, spark, config)
  result1 = future1.result()
  result2 = future2.result()

这将在一个进程中运行并使用多个进程，但实际的 Spark 作业将并行运行(Spark 远程执行释放 Python 的 GIL 类似于 IO 操作，尽管某些步骤可能需要它，例如当最终结果在 Spark 驱动程序中合并时)。