Должен ли тип переменных широковещательной передачи быть числом или строкой при попытке перезапустить задание с контрольной точки

Question

Когда я устанавливаю коллекцию как широковещательные переменные, она всегда возвращает мне ошибку сериализации, я уже пробовал Map, HashMap, Array, все не удалось

Answer 1

это известная ошибка Spark: https://issues.apache.org/jira/browse/SPARK-5206

вы можете использовать одноэлементный объект, чтобы каждый исполнитель загружал данные сам.Вы можете проверить https://github.com/apache/spark/blob/master/examples/src/main/java/org/apache/spark/examples/streaming/JavaRecoverableNetworkWordCount.java для полного примера:

class JavaWordBlacklist {

  private static volatile Broadcast<List<String>> instance = null;

  public static Broadcast<List<String>> getInstance(JavaSparkContext jsc) {
    if (instance == null) {
      synchronized (JavaWordBlacklist.class) {
        if (instance == null) {
          List<String> wordBlacklist = Arrays.asList("a", "b", "c");
          instance = jsc.broadcast(wordBlacklist);
        }
      }
    }
    return instance;
  }
}



public static void main(String[] args) throws Exception {
    ... 
    Function0<JavaStreamingContext> createContextFunc =
        () -> createContext(ip, port, checkpointDirectory, outputPath);

    JavaStreamingContext ssc =
      JavaStreamingContext.getOrCreate(checkpointDirectory, createContextFunc);
    ssc.start();
}

private static JavaStreamingContext createContext(String ip,
                                                    int port,
                                                    String checkpointDirectory,
                                                    String outputPath) {
    SparkConf sparkConf = new SparkConf().setAppName("JavaRecoverableNetworkWordCount");
    JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(1));
    ssc.checkpoint(checkpointDirectory);
    ...
    wordCounts.foreachRDD((rdd, time) -> {
      // Get or register the blacklist Broadcast
      Broadcast<List<String>> blacklist =
          JavaWordBlacklist.getInstance(new JavaSparkContext(rdd.context()))
     ...
     }
...

}