使用MRJob开发多步骤EMR/Spark作业：是否需为每个SparkStep新建SparkContext？

嘿，我来帮你理清这个问题！完全不需要为每个SparkStep创建新的SparkContext——这不仅是错误的做法，还会导致Spark抛出「Cannot run multiple SparkContexts at once」的异常，同时严重浪费集群资源。

核心原因： #

Spark的设计理念是一个应用对应一个SparkContext，它负责管理整个作业的集群资源、作业调度等核心逻辑。MRJob的多步骤Spark作业本质上是在同一个Spark应用内执行的不同计算阶段，所有步骤共享同一个Context才是正确的打开方式。

正确的实现方式： #

你应该使用SparkContext.getOrCreate()方法，它会自动检查当前是否已存在有效的SparkContext，有则直接复用，没有才会创建新的。这样既避免了重复创建的错误，又能保证所有步骤共享同一个Context。

给你修改后的示例代码：

from mrjob.job import MRJob
from pyspark import SparkContext

class MRSparkJob(MRJob):
    def spark_step1(self, input_path, output_path):
        # 复用已有的SparkContext，无需重新初始化
        sc = SparkContext.getOrCreate()
        # 这里写step1的业务逻辑
        rdd = sc.textFile(input_path)
        processed_rdd = rdd.filter(lambda line: len(line.strip()) > 0)
        processed_rdd.saveAsTextFile(output_path)

    def spark_step2(self, input_path, output_path):
        # 同样复用同一个SparkContext
        sc = SparkContext.getOrCreate()
        rdd = sc.textFile(input_path)
        # step2的业务逻辑，比如进行聚合计算
        result_rdd = rdd.map(lambda line: (line.split(',')[0], 1)).reduceByKey(lambda a, b: a + b)
        result_rdd.saveAsTextFile(output_path)

    def steps(self):
        return [
            self.spark_step(),
            self.spark_step()
        ]

if __name__ == '__main__':
    MRSparkJob.run()

额外注意事项： #

• 在EMR集群上运行时，MRJob会自动帮你完成SparkContext的初始化工作，你只需要在各个步骤中通过getOrCreate()获取即可。
• 不要尝试手动关闭SparkContext（比如调用sc.stop()），否则后续步骤会因为Context被销毁而无法执行。MRJob会在整个作业完成后自动清理资源。

内容的提问来源于stack exchange，提问作者vkc