Pyspark Luigi 多个工人问题答案

【问题标题】：Pyspark Luigi multiple workers issuePyspark Luigi 多个工人问题
【发布时间】：2021-05-18 06:28:00
【问题描述】：

我想使用 Luigi 工作流并行加载 spark 数据帧中的多个文件并将它们存储在字典中。加载所有文件后，我希望能够从主字典中访问这些数据帧，然后进行进一步处理。当我与一名工人一起运行 Luigi 时，此过程正在工作。如果与一名以上工人一起运行 Luigi，此变量在 main 方法中为空。

任何建议都会有所帮助。

 import Luigi
    from Luigi import LocalTarget
    
    from pyspark import SQLContext
    
    from src.etl.SparkAbstract import SparkAbstract
    from src.util.getSpark import  get_spark_session
    from  src.util import getSpark,read_json
    import configparser as cp
    import  datetime
    from src.input.InputCSVFileComponent import InputCSVFile
    import os
    from src.etl.Component import ComponentInfo
    
    class fileloadTask(luigi.Task):
    
        compinfo = luigi.Parameter()
    
        def output(self):
            return luigi.LocalTarget("src/workflow_output/"+str(datetime.date.today().isoformat() )+"-"+ str(self.compinfo.id)+".csv")
    
        def run(self):
    
            a = InputCSVFile(self.compinfo)  ##this class is responsible to return the object of  spark dataframe and put it in dictionary
            a.execute()
            with self.output().open('w') as f:
                f.write("done")
    
    class EnqueueTask(luigi.WrapperTask):
     compinfo = read_json.read_json_config('path to json file')
    
        def requires(self):
            folders = [
                comp.id for comp in list(self.compinfo) if comp.component_type == 'INPUTFILE'
            ]
            print(folders)
            newcominfo = []
            for index, objid in enumerate(folders):
                newcominfo.append(self.compinfo[index])
    
            for i in newcominfo:
                print(f" in compingo..{i.id}")
    
            callmethod = [fileloadTask(compinfo) for compinfo in newcominfo]
            print(callmethod)
    
            return callmethod
    
    class MainTask(luigi.WrapperTask):
    
        def requires(self):
            return EnqueueTask()
    
        def output(self):
            return luigi.LocalTarget("src/workflow_output/"+str(datetime.date.today().isoformat() )+"-"+ "maintask"+".csv")
    
        def run(self):
            print(f"printing mapdf..{SparkAbstract.mapDf}")
            res = not SparkAbstract.mapDf
            print("Is dictionary empty ? : " + str(res)) ####-------------> this is empty when workers > 1 ################
            for key, value in SparkAbstract.mapDf.items():
                print("prinitng from dict")
                print(key, value.show(10))
    
            with self.output().open('w') as f:
                f.write("done")
    
    """
    entry point for spark application
    """
    if __name__ == "__main__":
        luigi.build([MainTask()],workers=2,local_scheduler=True)

【问题讨论】：

标签： python apache-spark pyspark multiprocessing luigi

【解决方案1】：

每个工作人员都在自己的进程中运行。这意味着工作人员不能共享 python 对象（在这种情况下是您放置结果的字典）。

一般来说，luigi 最适合编排具有副作用的任务（例如写入文件等）。

如果您尝试并行化将数据加载到内存中的任务，我建议使用 dask 而不是 luigi。

【讨论】：