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

Posted 2023-04-13

【中文标题】导入pyspark ETL模块并使用python子进程作为子进程运行时出错

【英文标题】：Error while Importing pyspark ETL module and running as child process using pything subprocess

【发布时间】：2021-04-17 18:18:04

【问题描述】：

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

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

main.py

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

【参考方案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 驱动程序中合并时） .

【讨论】：

我引用了这个levelup.gitconnected.com/…和这个nurmatova.com/subprocesses-and-multithreading.html，了解到子进程用自己的内存运行子进程，这就是我想要的，在某些情况下，我必须按顺序运行子进程，而在其他情况下与它自己的记忆平行。如果我没记错的话，多处理使用相同的父进程内存，不是吗？在我的脚本中，在本地运行进程是否有问题？或不返回任何东西？我尝试返回一些虚拟值，即使那样我也得到了同样的错误

你错了 - 多进程正在运行具有自己内存的子进程，其方式与子进程相同。它们在这个意义上是相似的。但是 subprocess 用于运行带有参数的可执行文件（任何可执行文件 - 如 java 或 gcc 或 rsync），而 multiprocessing 是专门生成新的 python 进程来运行通过 pickle 序列化传递给它的 python 可调用对象，参数也传递给它pickle 序列化。

当您对 Python 使用表达式 subprocess.Popen(my_function(my_argument)) 时，它看起来与 any_other_callable(my_function(my_argument)) 没有什么不同 - 它会在本地计算 my_function(my_argument) 并尝试将其传递给 any_other_callable。在您的情况下，any_other_callable 是 subprocess.Popen，并且在由于本地运行 job_module1.run_etl_01_job(spark, config) 而返回它之后，您将其传递给 None。

如果你有函数 my_function 和参数 my_argument 并且想要生成一个新的 python 进程，将 my_function 应用到 my_argument 并返回结果 - 那正是 multiprocessing 的用途。

Python 的threading 模块在接口方面看起来与multiprocessing 相似，但它实际上是在python 进程中创建新线程而不是新的python 进程。因此，如果您使用threading - 那就是使用相同父进程内存的那个。问题是你可能会更好地使用它，因为它不需要序列化/反序列化你的 Spark 会话。