Исключительная структурированная потоковая передача ForeachWriter не может получить sparkContext

Question

Исключительная структурированная потоковая передача ForeachWriter не может получить sparkContext

Я читаю данные JSON из очереди Kafka с использованием структурированной потоковой передачи Spark, но мне нужно записать данные JSON в Elasticsearch.

Тем не менее, я не могу получить sparkContext внутри ForeachWriter конвертировать JSON в RDD. Это бросает NPE.

Как я могу получить SparkContext в Writer конвертировать JSON в RDD?

1

scala apache-spark structured-streaming elasticsearch-spark

Источник

user706454 22 фев '17 в 14:41

2 ответа

Другие вопросы по тегам scala apache-spark structured-streaming elasticsearch-spark

user1038826 22 фев '17 в 20:53 2017-02-22 20:53 · Answer 1 · 2017-02-22 20:53

Ты не можешь. Методы в ForeachWriter беги в исполнителях. Вы можете либо написать приемник Elasticsearch самостоятельно, либо вызвать необработанные API-интерфейсы Elasticsearch для записи данных.

2

Источник

user1038826 22 фев '17 в 20:53

user706454 22 фев '17 в 22:55 2017-02-22 22:55 · Answer 2 · 2017-02-22 22:55

Я решил проблему, получив экземпляр SparkContext внутри ForeachWriter

val writer = new ForeachWriter[CustomerData] {

  override def open(partitionId: Long, version: Long) = true
  override def process(value: CustomerData) = {
        val spark = SparkSession
        .builder()
       .getOrCreate()    //this works
       ...
  }
  override def close(errorOrNull: Throwable) = {}
}

PS: это, вероятно, создание новой SparkSession