scala insert to redis 给出任务不可序列化

Question

我有以下代码:-

case class event(imei: String, date: String, gpsdt: String,
             entrygpsdt: String,lastgpsdt: String) 

val result = rdd.map(row => {
val imei = row.getString(0)
val date = row.getString(1)
val gpsdt = row.getString(2)

event(imei, date, gpsdt, lastgpsdt ,"2018-04-06 10:10:10")
}).collect()

 val collection = sc.parallelize(result)
 collection.saveToCassandra("db", "table", SomeColumns("imei", "date", "gpsdt", "lastgpsdt", "dt")

这很好用。所以，现在我将这个结果值插入到 cassandra 中，但我也想将每个 rdd 的一部分插入到 Redis 中。当我尝试在循环中使用 redis 插入时，它给出了一个任务不可序列化的错误

我想要这样的东西:-

case class event(imei: String, date: String, gpsdt: String,
         entrygpsdt: String,lastgpsdt: String) 
val result = rdd.map(row => { 
val imei = row.getString(0)
val date = row.getString(1)
val gpsdt = row.getString(2)
val zscore = Calendar.getInstance().getTimeInMillis
val value = row.getString(0) + ',' + row.getString(2)
val key = row.getString(1)
client.zadd(key , zscore, value)
event(imei, date, gpsdt, lastgpsdt ,"2018-04-06 10:10:10")
}).collect()

val collection = sc.parallelize(result)
collection.saveToCassandra("db", "table", SomeColumns("imei", "date", "gpsdt", "lastgpsdt", "dt")

那么，我该怎么做，“客户端”是 Scala Redis 库的对象。谢谢，

Answer 1

因为没有人提供答案。我找到了适合我案例的解决方案。不知道这种方法是否好，但它对我有用。因此，想法是通过遍历 RDD 来收集数据。您将获得 Array[event] 的结果。所以，现在再次开始循环结果并将每一行插入 Redis。最后是 Cassandra 的“结果”。这个流程正在解决我一直在寻找的两个目的。谢谢，