Spark Context non è serializzabile.
Qualsiasi implementazione di ForeachWriter deve essere serializzabile perché ogni attività riceverà una nuova copia serializzata e deserializzata dell'oggetto fornito. Pertanto, si consiglia vivamente di eseguire qualsiasi inizializzazione per la scrittura di dati (ad esempio l'apertura di una connessione o l'avvio di una transazione) dopo che è stato chiamato il metodo open(...), il che significa che l'attività è pronta per generare dati.
Nel tuo codice, stai tentando di utilizzare il contesto spark all'interno del metodo di processo,
override def process(record: Row) = {
val stringHashRDD = sc.parallelize(Seq(("lastContact", record(1).toString)))
*sc.toRedisHASH(stringHashRDD, record(0).toString)(redisConfig)*
}
Per inviare dati a redis, devi creare la tua connessione e aprirla nel metodo aperto e quindi utilizzarla nel metodo di elaborazione.
Dai un'occhiata a come creare un pool di connessioni redis. https://github.com/RedisLabs/spark-redis/blob/master/src/main/scala/com/redislabs/provider/redis/ConnectionPool.scala
