如何在PySpark中处理数据之前在所有Spark worker上运行一个函数？

Question

我正在使用YARN在集群中运行Spark Streaming任务.集群中的每个节点都运行多个spark worker.在流式传输开始之前,我想在群集中所有节点上的所有工作程序上执行"设置"功能.

流式传输任务将传入的邮件分类为垃圾邮件或非垃圾邮件,但在此之前,它需要将最新的预先训练的模型从HDFS下载到本地磁盘,如此伪代码示例:

def fetch_models():
    if hadoop.version > local.version:
        hadoop.download()

我在SO上看过以下示例:

sc.parallelize().map(fetch_models)

但是在Spark 1.6中parallelize()需要使用一些数据,比如我现在正在做的这种糟糕的解决方法:

sc.parallelize(range(1, 1000)).map(fetch_models)

为了确保该函数在所有工作程序上运行,我将范围设置为1000.我还不确切地知道在运行时集群中有多少个工作程序.

我已经阅读了编程文档并无情地搜索了,但我似乎无法找到任何方法实际上只向所有工作人员分发任何没有任何数据的东西.

完成此初始化阶段后,流式传输任务与往常一样,对来自Kafka的传入数据进行操作.

我使用模型的方法是运行类似这样的函数:

spark_partitions = config.get(ConfigKeys.SPARK_PARTITIONS)
stream.union(*create_kafka_streams())\
    .repartition(spark_partitions)\
    .foreachRDD(lambda rdd: rdd.foreachPartition(lambda partition: spam.on_partition(config, partition)))

从理论上讲,我可以检查模型中的模型是否是最新的on_partition,尽管在每个批次上执行此操作会非常浪费.我想在Spark开始从Kafka检索批次之前这样做,因为从HDFS下载可能需要几分钟......

更新:

要明确:这不是关于如何分发文件或如何加载它们的问题,而是关于如何在不对任何数据进行操作的情况下对所有工作程序运行任意方法.

澄清当前实际加载模型的含义:

def on_partition(config, partition):
    if not MyClassifier.is_loaded():
        MyClassifier.load_models(config)

    handle_partition(config, partition)

虽然MyClassifier是这样的:

class MyClassifier:
    clf = None

    @staticmethod
    def is_loaded():
        return MyClassifier.clf is not None

    @staticmethod
    def load_models(config):
        MyClassifier.clf = load_from_file(config)

静态方法,因为PySpark似乎无法使用非静态方法序列化类(类的状态与另一个worker的关系无关).在这里,我们只需要调用load_models()一次,并在将来的所有批次中调用MyClassifier.clf.对于每个批次来说,这是不应该做的事情,这是一次性的事情.与使用fetch_models()从HDFS下载文件相同.

Answer 1

如果您只想在工作机器之间分发文件,最简单的方法是使用SparkFiles机制:

some_path = ...  # local file, a file in DFS, an HTTP, HTTPS or FTP URI.
sc.addFile(some_path)

并使用SparkFiles.get和标准IO工具在worker上检索它:

from pyspark import SparkFiles

with open(SparkFiles.get(some_path)) as fw:
    ... # Do something

如果要确保实际加载模型,最简单的方法是加载模块导入.假设config可以用于检索模型路径:

• model.py:

  from pyspark import SparkFiles
  
  config = ...
  class MyClassifier:
      clf = None
  
      @staticmethod
      def is_loaded():
          return MyClassifier.clf is not None
  
      @staticmethod
      def load_models(config):
          path = SparkFiles.get(config.get("model_file"))
          MyClassifier.clf = load_from_file(path)
  
  # Executed once per interpreter 
  MyClassifier.load_models(config)  
  

• main.py:

  from pyspark import SparkContext
  
  config = ...
  
  sc = SparkContext("local", "foo")
  
  # Executed before StreamingContext starts
  sc.addFile(config.get("model_file"))
  sc.addPyFile("model.py")
  
  import model
  
  ssc = ...
  stream = ...
  stream.map(model.MyClassifier.do_something).pprint()
  
  ssc.start()
  ssc.awaitTermination()
  

Answer 2

这是 Spark广播变量的典型用例。假设fetch_models返回模型而不是将它们保存在本地，您可以执行以下操作：

bc_models = sc.broadcast(fetch_models())

spark_partitions = config.get(ConfigKeys.SPARK_PARTITIONS)
stream.union(*create_kafka_streams())\
    .repartition(spark_partitions)\
    .foreachRDD(lambda rdd: rdd.foreachPartition(lambda partition: spam.on_partition(config, partition, bc_models.value)))

这确实假设您的模型适合内存、驱动程序和执行程序。

您可能会担心将模型从单个驱动程序广播到所有执行程序效率低下，但根据此分析，它使用的“高效广播算法”可以显着优于通过 HDFS 分发