Spark Structured streaming ForeachWriter не может получить sparkContext

Question

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

Однако я не могу получить sparkContext внутри ForeachWriter, чтобы преобразовать json в RDD. Это бросает NPE.

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

Answer 1

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

Answer 2

Я решил проблему, получив экземпляр 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