时间戳

首页 > 解决方案 > 导入 pyspark ETL 模块并使用 pything 子进程作为子进程运行时出错

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

问题描述

我正在尝试使用导入模块和子进程从一个 main.py python 脚本动态调用 pyspark 模块列表。我试图调用的子模块不返回任何内容，它只是执行其 ETL 操作。我希望我的 main.py 程序等到子进程完成。在下面的代码中，每次我尝试调用子进程时，都会出现错误“TypeError：'NoneType' object is not iterable”。另一个问题是，在启动 subprocess.Popen 之后，我认为流程将继续在 main.py 中继续到下一行，直到它到达 j1.wait()，但立即打印语句 (print("etl_01_job 正在运行") 没有执行，我错过了什么吗？

我用谷歌搜索并尝试了很多其他方法，但没有任何效果。谁能阐明我做错了什么？一旦我能够成功调用子进程，我必须根据子进程的返回码添加一些其他条件。但在这一点上，我想解决这个问题。谢谢

主文件

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

标签： pythonpyspark

解决方案

原因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)which 反过来会给您发布的错误。

您可能打算做的是使用multiprocessingpython 模块，如

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 驱动程序中合并时）。

推荐阅读