从可变长度 CSV 到对 RDD 的 Spark 转换

Question

我是 scala spark 的新手，我有一个 CSV 文件，如下所示。

R001, A, 10, C, 30, D, 50, X, 90
R002, E, 40, F, 70, G, 80, H, 90, J, 25 
R003, L, 30, M, 54, N, 67, O, 25, P, 85, Q, 100

如何将其转换为以行的第一个值为 Key 并跳过数值的 Pair RDD？

R001, A
R001, C
R001, D
R001, X
R002, E
R002, F
R002, G
R002, H
R002, J
R003, L
R003, M
R003, N
R003, O
R003, P
R003, Q

我试过了，看起来还不错，但我认为有更好的方法来做到这一点。

def isNumeric(str:String): Boolean = str.matches("[-+]?\\d+(\\.\\d+)?")

def process(field: Array[String]): Array[String] = { 
 val results = new Array[String](field.length)  
 for (i <- 1 to field.length-1) {
    if(!isNumeric(field(i).trim) && field(0)!=null && field(i)!=null)
     results(i)= field(0)+":"+field(i)
 }  
  results  
};

 def skipNulls(input : String) : String = {  
  if(input!=null && input.trim!="" && input.contains(":")) {
  var res = input.split(":")    
  res(0)+","+res(1)
  }  else {
    "null"
  }
};


val a= raw_csv.map(_.split(",")).flatMap(k => process(k))
val b= a.map(k => skipNulls(k))
val c = b.filter( x => x.contains("null")==false)
val d= c.toDF()
d.show
display(d)

Answer 1

看起来您对 Spark 部分的想法基本上是正确的（但看起来您实际上并没有达到您想要的 RDD？我假设您知道这一点），问题是我们是否可以清理process 和 skipNulls。因为你用的是flatMap，所以我想你应该可以过滤掉process中的坏点。

如果我理解正确，您将引入空值（然后稍后将其删除），因为您有这个固定长度的数组。但是，我们可以将“键”和“值”分开，过滤到非数字值，然后组装我们想要的对，而无需创建这个数组。比如：

def process(line: String): List[(String, String)] = {
  val key :: values = line.split(",").toList
  values.filterNot(isNumeric).map(key -> _) // equivalent to .map(x => (key, x))
}

这结合了您的前几个步骤，所以我的版本将是 raw_csv.flatMap(process) 以到达 RDD[(String, String)]。

拆分列表可以说有点神奇。以下方法也可以：

val elements = line.split(",")
val key = elements.head
val values = elements.tail

您可以将isNumeric 重写为：

def isNumeric(s: String): Boolean = Try(s.toFloat).isSuccess

对于这是否比正则表达式更好或更差，我没有强烈的意见。 Try 在scala.util 中。

Answer 2

如果您准备采用不同的方法，您可以通过首先将 csv 文本的每一行转换为两个字符串来简化解决方案：键和行的其余部分（从而规避可变长度问题）。

步骤：1) 通过在第一个 "," 处拆分，将源文本文件的每一行分成一个关键字符串和余数字符串数组，2) 使用 replaceAllIn() 删除余数字符串中的所有数字字段并将其拆分到一个数组中，并且，3) 将 (key, array) 元素展平为所需的键值对。

val numericField = """\s*[-+]?\d+(\.\d+)?\s*,?""".r

sc.textFile("/path/to/csvFile").
  map( _.split(",\\s*", 2) ).
  map{ case Array(key, remainder) =>
    (key, numericField.replaceAllIn(remainder, "").split(",\\s*")) }.
  flatMap{ case (k, arr) => arr.map( (k, _) ) }
// res1: org.apache.spark.rdd.RDD[(String, String)] = MapPartitionsRDD[21] ...

res1.collect
// res2: Array[(String, String)] = Array(
//   (R001,A), (R001,C), (R001,D), (R001,X),
//   (R002,E), (R002,F), (R002,G), (R002,H), (R002,J),
//   (R003,L), (R003,M), (R003,N), (R003,O), (R003,P), (R003,Q)
// )

Answer 3

hoyland 发布了答案。

我只是再次提供完整的代码

val raw_csv = sc.parallelize(Array(R001, A, 10, C, 30, D, 50, X, 90
R002, E, 40, F, 70, G, 80, H, 90, J, 25 
R003, L, 30, M, 54, N, 67, O, 25, P, 85, Q, 100))

import scala.util.{Try, Success, Failure}
def isNumeric(s: String): Boolean = Try(s.toFloat).isSuccess

def process(line: String): List[(String, String)] = {
  val key :: values = line.split(",").toList
  values.filterNot(isNumeric).map(key -> _) 
}

val processed = raw_csv.flatMap(k => process(k))
display(processed.toDF)

结果如下

R001, A
R001, C
R001, D
R001, X
R002, E
R002, F
R002, G
R002, H
R002, J
R003, L
R003, M
R003, N
R003, O
R003, P
R003, Q