Spark Streaming: сделать обновляемый набор результатов доступным для логики обработки потока

Question

В приложении Spark Streaming есть внешний источник данных (реляционная база данных), который мне нужно запрашивать каждые 10 минут и сделать результаты доступными для моего конвейера обработки потока.

Я не совсем понимаю, как правильно это сделать. Accumulators будут добавлять только (как описано в документации), но я нашел это:

/** 
    * Set the accumulator's value; only allowed on master 
    */ 
    def setValue(newValue: R) { 
    this.value = newValue 
    } 

и Broadcast variables только однократной записи https://spark.apache.org/docs/1.6.2/programming-guide.html#broadcast-variables

аспект планирования также не ясно для меня.

Есть ли способ сделать обновленный набор результатов доступным для логики обработки потока?

PS Это, кажется, очень похоже на то, что мне нужно How can I update a broadcast variable in spark streaming?

Answer 1

Я делаю сохранить в Java и работает достаточно. Существует аналогичный ответ здесь Updating a global variable periodically in Spark, а также упоминается в вопросе

public Broadcast<Map<String, List<String>>> updateBroadcastVariable(
    SparkContext sparkContext, DatabaseService databaseService) { 
Date d = Calendar.getInstance().getTime(); 
long diff = d.getTime()-mysqlLastUpdatedAt.getTime(); 
if (updatedVar == null || diff > 60000) { 
    if (var != null) 
    updatedVar.unpersist(); 
    mysqlLastUpdatedAt = new Date(System.currentTimeMillis()); 
    Map<String, List<String>> map = new LinkedHashMap<>(); 

    List<String> dbData = databaseService.refreshData(JavaSparkContext.fromSparkContext(sparkContext)); 
    } 
    updatedVar = JavaSparkContext.fromSparkContext(sparkContext).broadcast(dbData); 
} 
return updatedVar; 
}