如何更新火花流中的广播变量？

Question

我相信，我有一个相对常见的火花流用例：

我有一个对象流，我想根据一些参考数据进行过滤

最初，我认为使用广播变量来实现这将是一件非常简单的事情：

public void startSparkEngine {
    Broadcast<ReferenceData> refdataBroadcast
      = sparkContext.broadcast(getRefData());

    final JavaDStream<MyObject> filteredStream = objectStream.filter(obj -> {
        final ReferenceData refData = refdataBroadcast.getValue();
        return obj.getField().equals(refData.getField());
    }

    filteredStream.foreachRDD(rdd -> {
        rdd.foreach(obj -> {
            // Final processing of filtered objects
        });
        return null;
    });
}

但是，尽管很少，我的参考数据会定期更改

我的印象是我可以在驱动程序上修改和重新广播我的变量，它会传播到每个工人，但是 Broadcast 对象不是 Serializable并且需要是final。

我有什么选择？我能想到的三个解决方案是：

1. 将参考数据查找移动到forEachPartition 或forEachRdd，使其完全驻留在工作人员身上。然而，参考数据存在于 REST API 中，因此我还需要以某种方式存储一个计时器/计数器，以停止远程访问流中的每个元素。

2. 每次 refdata 更改时都使用新的广播变量重新启动 Spark 上下文。

3. 将参考数据转换为RDD，然后join 以我现在正在流式传输Pair<MyObject, RefData> 的方式，尽管这会将参考数据与每个对象一起传送.

Answer 1

通过@Rohan Aletty 扩展答案。这是一个基于 ttl 刷新广播变量的 BroadcastWrapper 的示例代码

public class BroadcastWrapper {

    private Broadcast<ReferenceData> broadcastVar;
    private Date lastUpdatedAt = Calendar.getInstance().getTime();

    private static BroadcastWrapper obj = new BroadcastWrapper();

    private BroadcastWrapper(){}

    public static BroadcastWrapper getInstance() {
        return obj;
    }

    public JavaSparkContext getSparkContext(SparkContext sc) {
       JavaSparkContext jsc = JavaSparkContext.fromSparkContext(sc);
       return jsc;
    }

    public Broadcast<ReferenceData> updateAndGet(SparkContext sparkContext){
        Date currentDate = Calendar.getInstance().getTime();
        long diff = currentDate.getTime()-lastUpdatedAt.getTime();
        if (var == null || diff > 60000) { //Lets say we want to refresh every 1 min = 60000 ms
            if (var != null)
               var.unpersist();
            lastUpdatedAt = new Date(System.currentTimeMillis());

            //Your logic to refresh
            ReferenceData data = getRefData();

            var = getSparkContext(sparkContext).broadcast(data);
       }
       return var;
   }
}

您的代码如下所示：

public void startSparkEngine() {

    final JavaDStream<MyObject> filteredStream = objectStream.transform(stream -> {
        Broadcast<ReferenceData> refdataBroadcast = BroadcastWrapper.getInstance().updateAndGet(stream.context());

        stream.filter(obj -> obj.getField().equals(refdataBroadcast.getValue().getField()));
    });

    filteredStream.foreachRDD(rdd -> {
        rdd.foreach(obj -> {
        // Final processing of filtered objects
        });
        return null;
    });
}

这对我也适用于多集群。 希望这会有所帮助

Answer 2

最近遇到了这个问题。认为它可能对 scala 用户有帮助..

Scala 处理BroadCastWrapper 的方式如下例所示。

import java.io.{ ObjectInputStream, ObjectOutputStream }
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.streaming.StreamingContext
import scala.reflect.ClassTag

/* wrapper lets us update brodcast variables within DStreams' foreachRDD
 without running into serialization issues */
case class BroadcastWrapper[T: ClassTag](
 @transient private val ssc: StreamingContext,
  @transient private val _v: T) {

  @transient private var v = ssc.sparkContext.broadcast(_v)

  def update(newValue: T, blocking: Boolean = false): Unit = {

    v.unpersist(blocking)
    v = ssc.sparkContext.broadcast(newValue)
  }

  def value: T = v.value

  private def writeObject(out: ObjectOutputStream): Unit = {
    out.writeObject(v)
  }

  private def readObject(in: ObjectInputStream): Unit = {
    v = in.readObject().asInstanceOf[Broadcast[T]]
  }
}

每次你需要调用更新函数来获取新的广播变量。

Answer 3

几乎每个处理流式应用程序的人都需要一种方法将参考数据（来自数据库、文件等）编织（过滤、查找等）到流式数据中。我们有整个两部分的部分解决方案

1. 查找要在流式操作中使用的参考数据

   • 创建具有所需缓存 TTL 的 CacheLookup 对象
   • 将其包装在广播中
   • 使用 CacheLookup 作为流逻辑的一部分

在大多数情况下，这都可以正常工作，但以下情况除外

1. 更新参考数据

   尽管这些线程中提出了建议，但没有确定的方法来实现这一点，即：杀死以前的广播变量并创建新的。多个未知数，例如这些操作之间的预期。

这是一个常见的需求，如果有一种方法可以将信息发送到广播变量通知更新，那将会有所帮助。这样，可以使“CacheLookup”中的本地缓存无效

问题的第二部分仍未解决。如果有任何可行的方法，我会很感兴趣

Answer 4

不确定您是否已经尝试过，但我认为可以在不关闭SparkContext 的情况下更新广播变量。通过使用unpersist() 方法，广播变量的副本在每个执行程序上被删除，并且需要重新广播才能再次访问该变量。对于您的用例，当您想更新广播时，您可以：

1. 等待您的执行者完成当前的一系列数据

2. 解除广播变量

3. 更新广播变量

4. 重播以将新的参考数据发送给执行者

我从this post 大量绘制，但最后一个回复的人声称它已在本地工作。重要的是要注意，您可能希望在 unpersist 上将阻塞设置为 true，以便您可以确保执行程序删除了旧数据（因此在下一次迭代中不会再次读取过时的值）。

Answer 5

我以不同的方式做到了。

我创建了一个广播变量，并每 5 分钟在驱动程序的不同线程中更新它。

  var broadcastValue: Broadcast[Set[String]] = spark.sparkContext.broadcast(calculateValue())

  def runScheduledThreadToUpdateBroadcastVariable(): Unit = {
    val updateTask = new Runnable {
      def run() = {
        broadcastValue.unpersist(blocking = false)
        broadcastValue = spark.sparkContext.broadcast(calculateValue())
      }
    }

    val executor = new ScheduledThreadPoolExecutor(1)
    executor.scheduleAtFixedRate(updateTask, 1, 5, TimeUnit.MINUTES)
  }

Answer 6

实现的最简单方法，下面的代码读取每个批次的维度数据文件夹，但请记住新的维度数据值（在我的例子中为国家名称）必须是一个新文件。

package com.databroccoli.streaming.dimensionupateinstreaming

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{DataFrame, ForeachWriter, Row, SparkSession}
import org.apache.spark.sql.functions.{broadcast, expr}
import org.apache.spark.sql.types.{StringType, StructField, StructType, TimestampType}

object RefreshDimensionInStreaming {

  def main(args: Array[String]) = {

    @transient lazy val logger: Logger = Logger.getLogger(getClass.getName)

    Logger.getLogger("akka").setLevel(Level.WARN)
    Logger.getLogger("org").setLevel(Level.ERROR)
    Logger.getLogger("com.amazonaws").setLevel(Level.ERROR)
    Logger.getLogger("com.amazon.ws").setLevel(Level.ERROR)
    Logger.getLogger("io.netty").setLevel(Level.ERROR)

    val spark = SparkSession
      .builder()
      .master("local")
      .getOrCreate()

    val schemaUntyped1 = StructType(
      Array(
        StructField("id", StringType),
        StructField("customrid", StringType),
        StructField("customername", StringType),
        StructField("countrycode", StringType),
        StructField("timestamp_column_fin_1", TimestampType)
      ))

    val schemaUntyped2 = StructType(
      Array(
        StructField("id", StringType),
        StructField("countrycode", StringType),
        StructField("countryname", StringType),
        StructField("timestamp_column_fin_2", TimestampType)
      ))

    val factDf1 = spark.readStream
      .schema(schemaUntyped1)
      .option("header", "true")
      .csv("src/main/resources/broadcasttest/fact")

    var countryDf: Option[DataFrame] = None: Option[DataFrame]

    def updateDimensionDf() = {
      val dimDf2 = spark.read
        .schema(schemaUntyped2)
        .option("header", "true")
        .csv("src/main/resources/broadcasttest/dimension")

      if (countryDf != None) {
        countryDf.get.unpersist()
      }

      countryDf = Some(
        dimDf2
          .withColumnRenamed("id", "id_2")
          .withColumnRenamed("countrycode", "countrycode_2"))

      countryDf.get.show()
    }

    factDf1.writeStream
      .outputMode("append")
      .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
        batchDF.show(10)

        updateDimensionDf()

        batchDF
          .join(
            countryDf.get,
            expr(
              """
      countrycode_2 = countrycode 
      """
            ),
            "leftOuter"
          )
          .show

      }
      .start()
      .awaitTermination()

  }

}