テキストファイルをSpark Contextに読み込む際にフィールドが欠落している行をスキップ

Question

Spark Contextにタブ区切りファイルをロードする必要があります。しかし、いくつかのフィールドには値がありません。これらの行を除外する必要があります。私は次のコードを使用しています。ただし、フィールドが完全に欠落している場合（行のタブが1つ少ないなど）、このコードは例外をスローします。これを達成するためのよりよい方法は何ですか？

val RDD = sc.textFile("file.txt").map(_.split("\t")) 
.filter(_(0).nonEmpty) 
.filter(_(1).nonEmpty) 
.filter(_(2).nonEmpty) 
.filter(_(3).nonEmpty) 
.filter(_(4).nonEmpty) 
.filter(_(5).nonEmpty) 

Answer 1

私は、大規模なデータセットのためにうまく動作するようにこれを発見した：

val allRecords: RDD[Either[(String, String, String, String), Array[String]]] = sc.textFile(s"file.txt") 
    .map(x=>x.split("\t")) 
    .map { 
    case Array(name, address, phone, country) => Left(name, address, phone, country) 
    case badArray => Right(badArray) 
    } 

val goodRecords = allRecords.collect{ case Left(r) => r } 

Answer 2

まず第一に、このことを考慮してください。

scala> "1\t2\t\t4\t".split("\t") 
res0: Array[String] = Array(1, 2, "", 4) 

scala> "1\t2\t\t4\t".split("\t", -1) 
res1: Array[String] = Array(1, 2, "", 4, "") 

2番目の引数として負の整数を渡すには、末尾の空要素が失われないことを保証します。

val RDD = sc.textFile("file.txt").map(_.split("\t", -1)) 
    .filter(array => array.forall(elem => elem.nonEmpty)) 

Answer 3

あなたはデータフレームとして、あなたのファイルを読み込み、DataFrameNaFunctions

val df = sqlContext.read.format("com.databricks.spark.csv").option("delimiter", "\t").load("file.txt") 
val cleanDF = df.na.drop() 

Hereは火花防止へのリンクである使用することができます。

は、以下のことを試してみてください（これは私の解決策の正しさのために不可欠です）ちょうどの場合のCSVライブラリ。