spark rdd中数据出现分隔符如何处理

Question

使用 spark RDD 加载文件时，如果我的分隔符出现在数据中，如何处理。

我的数据如下所示：

NAME|AGE|DEP
Suresh|32|BSC
"Sathish|Kannan"|30|BE

如何将此列转换为如下所示的 3 列。

NAME     AGE     DEP
suresh     32      Bsc
Sathish|Kannan      30     BE

请参考我尝试加载数据的方式。

scala> val rdd = sc.textFile("file:///test/Sample_dep_20.txt",2)
rdd: org.apache.spark.rdd.RDD[String] = hdfs://Hive/Sample_dep_20.txt MapPartitionsRDD[1] at textFile at <console>:27


rdd.collect.foreach(println)

101|"Sathish|Kannan"|BSC
102|Suresh|DEP


scala> val rdd2=rdd.map(x=>x.split("\""))
rdd2: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[2] at map at <console>:29

scala> val rdd3=rdd2.map(x=>
     | {
     | var strarr = scala.collection.mutable.ArrayBuffer[String]()
     | for(v<-x)
     | {
     | if(v.startsWith("\"") && v.endsWith("\""))
     |   strarr +=v.replace("\"","")
     | else if(v.contains(","))
     |   strarr ++=v.split(",")
     | else
     |   strarr +=v
     | }
     | strarr
     | }
     | )
rdd3: org.apache.spark.rdd.RDD[scala.collection.mutable.ArrayBuffer[String]] = MapPartitionsRDD[3] at map at <console>:31

scala> rdd3.collect.foreach(println)
ArrayBuffer(101|, Sathish|Kannan, |BSC)
ArrayBuffer(102|Suresh|DEP)

Answer 1

也许您需要将 " 明确定义为引号字符（默认情况下 csv reader 但在您的情况下可能不需要？）。因此，在读取 .csv 文件时将 .option("quote","\"") 添加到选项应该有效。

scala> val inputds = Seq("Suresh|32|BSC","\"Satish|Kannan\"|30|BE").toDS()
inputds: org.apache.spark.sql.Dataset[String] = [value: string]

scala> val outputdf = spark.read.option("header",false).option("delimiter","|").option("quote","\"").csv(inputds)
outputdf: org.apache.spark.sql.DataFrame = [_c0: string, _c1: string ... 1 more field]

scala> outputdf.show(false)
+-------------+---+---+
|_c0          |_c1|_c2|
+-------------+---+---+
|Suresh       |32 |BSC|
|Satish|Kannan|30 |BE |
+-------------+---+---+

定义使得 DataFrameReader 忽略引号字符串中的定界符，参见 Spark API 文档 here.

编辑

如果你想玩得开心并且仍然使用普通的 RDD，那么尝试像这样修改你的 split() 函数：

val rdd2=rdd.map(x=>x.split("\|(?=([^\"]*\"[^\"]*\")*[^\"]*$)"))

它使用正 look-ahead 来忽略引号内的 | 定界符，并避免您在第二个 .map.

中进行字符串操作

spark rdd中数据出现分隔符如何处理

How to handle if delimiter appears in data in spark rdd

scala

apache-spark

rdd