Spark Scala Dataframe Konvertierung

Question

Ich habe eine Textdatei, die mit ~ begrenzt ist, muss ich einige Parsing vor der Konvertierung in einen Datenrahmen durchführen. Der Code liest eine Textdatei ein, während RDD [String] ein Parsing durchführt. Dann konvertiert es zu RDD [Row]. Dann mit dem Schema erstellt einen Datenrahmen.

Also unten ist der folgende Code, den ich habe. Es funktioniert, aber das Problem ist das tatsächliche Schema ist 400 Felder lang. Ich habe mich gefragt, ob es einen einfacheren Weg gibt, als Attribute (1), Attribute (2), Attribute (3) usw. einzugeben.

Ich bin derzeit auf Spark 1.6. CDH 5.2.2

Beispiel Eingabe:

20161481132310 ~  ~"This" is a comma 10 

Aktuelle Code:

val schema_1 = StructType(Array(
StructField("EXAMPLE_1", StringType, true), 
StructField("EXAMPLE_2", StringType, true), 
StructField("EXAMPLE_3", StringType, true))) 

val rdd = sc.textFile("example.txt") 
val rdd_truncate = rdd.map(_.split("~").map(_.trim).mkString("~")) 
val row_final = rdd_truncate 
    .map(_.split("~")) 
    .map(attributes => Row(attributes(0), 
    attributes(1), 
    attributes(2))) 

val df = sqlContext.createDataFrame(row_final, schema_1) 

Basierend auf Vorschlag, den ich für folgende geändert. Es funktioniert, außer in Anführungszeichen. Das "Dies" in der Eingabe wird fehlschlagen. Irgendwelche Vorschläge?

val df = sqlContext.read 
     .format("com.databricks.spark.csv") 
     .option("delimiter","~") 
     .schema(schema) 
     .load("example.txt") 
val df_final = df.select(df.columns.map(c =>trim(col(c)).alias(c)): _*) 

Answer 1

Verwenden Sie einfach Standard-CSV-Leser:

spark.read.schema(schema).option("delimiter", "~").csv("example.txt") 

Wenn Sie Felder verwenden nur trimmen wollen select:

import org.apache.spark.sql.functions.{col, trim} 

df.select(df.columns.map(c => trim(col(c)).alias(c)): _*) 

Wenn Sie Spark-1.x verwenden, können Sie spark-csv verwenden:

sqlContext.read 
    .format("com.databricks.spark.csv") 
    .schema(schema) 
    .option("delimiter", "~") 
    .load("example.txt") 

Wenn dies aus irgendeinem Grund nicht ausreicht, können Sie Row.fromSeq:

Row.fromSeq(line.split("~").take(3)) 

verwenden