Spark Context ist nicht serialisierbar.
Jede Implementierung von ForeachWriter muss serialisierbar sein, da jede Aufgabe eine neue serialisierte/deserialisierte Kopie des bereitgestellten Objekts erhält. Daher wird dringend empfohlen, dass jede Initialisierung zum Schreiben von Daten (z. B. das Öffnen einer Verbindung oder das Starten einer Transaktion) nach dem Aufruf der open(...)-Methode erfolgt, was bedeutet, dass die Task bereit ist, Daten zu generieren.
In Ihrem Code versuchen Sie, den Spark-Kontext innerhalb der Prozessmethode zu verwenden,
override def process(record: Row) = {
val stringHashRDD = sc.parallelize(Seq(("lastContact", record(1).toString)))
*sc.toRedisHASH(stringHashRDD, record(0).toString)(redisConfig)*
}
Um Daten an redis zu senden, müssen Sie Ihre eigene Verbindung erstellen und sie in der open-Methode öffnen und dann in der process-Methode verwenden.
Sehen Sie sich an, wie Sie einen Redis-Verbindungspool erstellen. https://github.com/RedisLabs/spark-redis/blob/master/src/main/scala/com/redislabs/provider/redis/ConnectionPool.scala
