Как я могу сделать DStream после использования foreachrdd ()? [Дубликат]

Подход к внешним вызовам с внешней массой - это прямое преобразование RDD в DStream на уровне раздела.

Шаблон выглядит следующим образом:

val transformedStream = dstream.transform{rdd => 
    rdd.mapPartitions{iterator => 
      val externalService = Service.instance() // point to reserve local resources or make server connections.
      val data = iterator.toList // to act in bulk. Need to tune partitioning to avoid huge data loads at this level
      val resultCollection = externalService(data)
      resultCollection.iterator
    }
 }

Этот процесс подхода каждый раздел базового RDD параллельно с использованием ресурсов, доступных в кластере. Обратите внимание, что подключение к внешней системе должно быть создано для каждого раздела (а не для каждого элемента).