火花：向csv文件添加列名失败

Question

16777216    16777471        -33.4940    143.2104
16777472    16778239    Fuzhou  26.0614 119.3061

然后我跑：

sc.textFile("path/to/a.txt").map(line => line.split("\t")).toDF("startIP", "endIP", "City", "Longitude", "Latitude")

然后我得到：

IllegalArgumentException：需求失败：列数不匹配。旧列名（1）：值新列名（5）：startIP,endIP,City,Longitude,Latitude at scala.predef$.require(predef.scala:224)at org.apache.spark.sql.dataset.todf（dataset.scala:376)at org.apache.spark.sql.datasetholder.todf（datasetholder.scala:40)...47省略

res.map(line => line.split("\t")).take(2)

rdd: Array[Array[String]] = Array(Array(16777216, 16777471, "", -33.4940, 143.2104), Array(16777472, 16778239, Fuzhou, 26.0614, 119.3061))

这里怎么了？

Answer 1

正如@user7881163所指出的，发生错误是因为您的split生成单个列，其值（因此由Spark给出的value名称）是由split生成的标记数组。

但是，根据@Zero323的注释，如果您正在大规模操作，请确保使用collect@user7881163使用的版本（接受部分函数的版本），因为另一个更常用的collect将把所有数据转移到驱动程序中，并淹没该机器。如果你没有大规模操作，为什么要使用Spark呢？

这是一种稍微不同的方法，也允许丢失城市数据：

sc.textFile("path/to/a.txt")
  .map(_.split("\t"))
  .map {
      case Array(startIP, endIP, city, longitude, latitude) => (startIP, endIP, Some(city), longitude, latitude)
      case Array(startIP, endIP, longitude, latitude) => (startIP, endIP, None, longitude, latitude)
  }.toDF("startIP", "endIP", "City", "Longitude", "Latitude")