从 Spark 数据帧的行创建对象的规范方法是什么？

问题描述

我使用的是 Apache Zeppelin (0.9.0) 和 Scala (2.11.12)。我想从数据框中提取一些数据并将其存储到 InfluxDB，稍后在 Grafana 中进行可视化，但无法弄清楚。我正在尝试使用 foreach 循环的幼稚方法。这个想法是遍历所有行，提取我需要的列，创建一个 Point 对象（来自 this InfluxDB client library），然后将其发送到 InfluxDB 或将其添加到列表中，然后批量发送所有点，之后循环。

数据框如下所示：

+---------+---------+-------------+-----+
|timestamp|sessionId|eventDuration|speed|
+---------+---------+-------------+-----+
|        1|     ses1|          0.0|   50|
|        2|     ses1|          1.0|   50|
|        3|     ses1|          2.0|   50|

我已尝试执行上述操作：

import scala.collection.mutable.ListBuffer
import spark.implicits._
import org.apache.spark.sql._
import com.paulgoldbaum.influxdbclient._
import scala.concurrent.ExecutionContext.Implicits.global

val influxdb = InfluxDB.connect("172.17.0.4",8086)
val database = influxdb.selectDatabase("test")

var influxData = new ListBuffer[Point]()
dfAnalyseReport.foreach(row => 
    {
        val point = Point("acceleration")
                    .addTag("speedBin",row.getLong(3).toString)
                    .addField("eventDuration",row.getDouble(2))
        influxData += point
    }
)
val influxDataList = influxData.toList
database.bulkWrite(influxDataList)

我在这里得到的唯一信息是一个没有附加信息的神秘 java.lang.classCastException，无论是在笔记本输出还是 Zeppelin Docker 容器的日志中。该错误似乎在 foreach 中的某个地方，即使我注释掉最后两行也会出现该错误。

我也尝试改编 this answer 中的方法 1，为列使用案例类，但无济于事。我让它运行没有错误，但结果列表是空的。不幸的是，我删除了该尝试。如有必要，我可以重建它，但我已经在这方面花费了太多时间，我相当肯定我对如何做到这一点存在一些根本性的误解。

另一个问题：我还尝试在构建时将每个 Point 写入数据库（而不是批量写入）。唯一的区别是，我没有附加到 ListBuffer，而是执行了 database.write(point) 操作。当在循环外用一个虚拟点完成时，它通过没有问题 - 数据最终在 InfluxDB 中 - 但在循环内它导致 org.apache.spark.SparkException: Task not serializable

有人能以正确的方式指出我吗？我应该如何解决这个问题？

解决方法

我会用 RDD map 方法来做，并将结果收集到一个列表中：

val influxDataList = dfAnalyseReport.rdd.map(
    row => Point("acceleration")
           .addTag("speedBin",row.getInt(3).toString)
           .addField("eventDuration",row.getDouble(2))
).collect.toList