如何解析一些（子）字段为空的分隔字段？

Question

我在 spark-shell 中使用 Spark 2.1.1 和 Scala 2.11.8。

我的输入数据集类似于：

2017-06-18 00:00:00 , 1497769200 , z287570731_serv80i:7:175 , 5:Re

2017-06-18 00:00:00 , 1497769200 , p286274731_serv80i:6:100 , 138 

2017-06-18 00:00:00 , 1497769200 , t219420679_serv37i:2:50 , 5

2017-06-18 00:00:00 , 1497769200 , v290380588_serv81i:12:800 , 144:Jo

2017-06-18 00:00:00 , 1497769200 , z292902510_serv83i:4:45 , 5:Re

2017-06-18 00:00:00 , 1497769200 , v205454093_serv75i:5:70 , 50:AK

它保存为 CSV 文件，使用 sc.textFile("input path") 读取

经过几次转换，这是我拥有的 RDD 的输出：

(String, String) = ("Re ",7)

我通过执行得到这个

val tid = read_file.map { line =>
  val arr = line.split(",")
  (arr(3).split(":")(1), arr(2).split(":")(1))
}

我的输入 RDD 是：

( z287570731_serv80i:7:175 , 5:Re )

( p286274731_serv80i:6:100 , 138 )

( t219420679_serv37i:2:50 , 5 )

( v290380588_serv81i:12:800 , 144:Jo )

( z292902510_serv83i:4:45 , 5:Re )

可以观察到，在第一个条目第 2 列中，我有

5:Re 

我得到了输出

("Re ",7)

但是当我到达第二行时，根据格式，第 2 列是 138，应该是

138:null 

但在执行时给出 ArrayIndexOutOfBoundsException

tid.collect()

如何更正此问题，以便在第二行和第三行分别以 138 和 5 显示 null？我试着这样做：

tid.filter(x => x._1 != null )

Answer 1

问题是您希望该位置至少有两个部分，而您可能只有一个。

以下是导致问题的行。

{var arr = line.split(","); (arr(3).split(":")(1),arr(2).split(":")(1))});

在您执行line.split(",") 之后，您然后是arr(3).split(":")(1) 和arr(2).split(":")(1)。

对格式肯定有太多假设，被缺失值打败了。

但在执行时给出 ArrayIndexOutOfBoundsException

这是因为您访问 3 和 2 元素但只有 2 个 (!)

scala> sc.textFile("input.csv").
  map { line => line.split(",").toSeq }.
  foreach(println)
WrappedArray(( z287570731_serv80i:7:175i ,  5:Re ))
WrappedArray(( p286274731_serv80i:6:100 ,  138 ))

这个问题几乎与 Spark 无关。这是一个常规的 Scala 问题，其中数据不在您期望的位置。

scala> val arr = "hello,world".split(",")
arr: Array[String] = Array(hello, world)

请注意，上面只是一个纯 Scala。

解决方案 1 - Spark Core 的 RDD

给定以下数据集...

2017-06-18 00:00:00 , 1497769200 , z287570731_serv80i:7:175 , 5:Re
2017-06-18 00:00:00 , 1497769200 , p286274731_serv80i:6:100 , 138 
2017-06-18 00:00:00 , 1497769200 , t219420679_serv37i:2:50 , 5
2017-06-18 00:00:00 , 1497769200 , v290380588_serv81i:12:800 , 144:Jo
2017-06-18 00:00:00 , 1497769200 , z292902510_serv83i:4:45 , 5:Re
2017-06-18 00:00:00 , 1497769200 , v205454093_serv75i:5:70 , 50:AK

...我会做以下事情：

val solution = sc.textFile("input.csv").
  map { line => line.split(",") }.
  map { case Array(_, _, third, fourth) => (third, fourth) }.
  map { case (third, fourth) =>
    val Array(_, a @ _*) = fourth.split(":")
    val Array(_, right, _) = third.split(":")
    (a.headOption.orNull, right)
  }
scala> solution.foreach(println)
(Re,7)
(null,6)
(Re,4)
(null,2)
(AK,5)
(Jo,12)

解决方案 2 - Spark SQL 的数据帧

我强烈建议使用 Spark SQL 进行此类数据转换。正如您所说，您是 Spark 的新手，所以为什么不从正确的地方开始，这正是 Spark SQL。

val solution = spark.
  read.
  csv("input.csv").
  select($"_c2" as "third", $"_c3" as "fourth").
  withColumn("a", split($"fourth", ":")).
  withColumn("left", $"a"(1)).
  withColumn("right", split($"third", ":")(1)).
  select("left", "right")
scala> solution.show(false)
+----+-----+
|left|right|
+----+-----+
|Re  |7    |
|null|6    |
|null|2    |
|Jo  |12   |
|Re  |4    |
|AK  |5    |
+----+-----+

Answer 2

ArrayIndexOutOfBounds 正在发生，因为如果元组的第二个元素中不存在 :，则该元素将不存在。 您可以检查: 是否存在于每个元组的第二个元素中。然后使用 map 为您提供一个中间 RDD，您可以在该 RDD 上运行当前查询。

val rdd = sc.parallelize(Array(
    ( "z287570731_serv80i:7:175" , "5:Re" ),
    ( "p286274731_serv80i:6:100" , "138" ),
    ( "t219420679_serv37i:2:50" , "5" ),
    ( "v290380588_serv81i:12:800" , "144:Jo" ),
    ( "z292902510_serv83i:4:45" , "5:Re" )))


rdd.map { x =>
    val idx = x._2.lastIndexOf(":")
    if(idx == -1) (x._1, x._2+":null")
    else (x._1, x._2)
}

显然有更好的（更少的代码行）方法来完成您想要完成的任务，但作为初学者，最好将每个步骤安排在一个命令中，这样就很容易阅读和理解，特别是在您使用 scala 时只需一行代码即可阻止全球变暖。

Answer 3

如果您的数据在文件中如下所示

( z287570731_serv80i:7:175 , 5:Re )
( p286274731_serv80i:6:100 , 138 )
( t219420679_serv37i:2:50 , 5 )
( v290380588_serv81i:12:800 , 144:Jo )
( z292902510_serv83i:4:45 , 5:Re )

然后就可以使用了

val tid = sc.textFile("path to the input file")
  .map(line => line.split(","))
  .map(array => {
    if (array(1).contains(":")) (array(1).split(":")(1).replace(")", "").trim, array(0).split(":")(1))
    else (null, array(0).split(":")(1))
  })
tid.foreach(println)

这应该给你输出

(Re,7)
(null,6)
(null,2)
(Jo,12)
(Re,4)

但是如果你有数据

2017-06-18 00:00:00 , 1497769200 , z287570731_serv80i:7:175 , 5:Re
2017-06-18 00:00:00 , 1497769200 , p286274731_serv80i:6:100 , 138
2017-06-18 00:00:00 , 1497769200 , t219420679_serv37i:2:50 , 5
2017-06-18 00:00:00 , 1497769200 , v290380588_serv81i:12:800 , 144:Jo
2017-06-18 00:00:00 , 1497769200 , z292902510_serv83i:4:45 , 5:Re
2017-06-18 00:00:00 , 1497769200 , v205454093_serv75i:5:70 , 50:AK
2017-06-18 00:00:00 , 1497769200 , z287096299_serv80i:19:15000 , 39:Re

那你需要做的

val tid = sc.textFile("path to the input file")
  .map(line => line.split(","))
  .map(array => {
    if (array(3).contains(":")) (array(3).split(":")(1).replace(")", "").trim, array(2).split(":")(1))
    else (null, array(2).split(":")(1))
  })
tid.foreach(println)

你应该有输出

(Re,7)
(null,6)
(null,2)
(Jo,12)
(Re,4)
(AK,5)
(Re,19)