Ersetzen neue Zeile (\n) Zeichen in csv-Datei - spark-scala

Nur um das problem zu veranschaulichen habe ich ein testset csv-Datei. Aber im real-case-Szenario, das problem zu behandeln, mehr als ein TeraByte Daten.

Ich habe eine CSV Datei, wo die Spalten sind von Anführungszeichen eingeschlossen("col1"). Aber wenn das importieren der Daten getan wurde. Eine Spalte enthält neue-Zeile-Zeichen(\n). Dies führt mich zu viele Probleme, wenn ich wollen speichern Sie Sie als Hive-Tabellen.

Meine Idee war, ersetzen Sie die \n-Zeichen mit der pipe "|" in der Funke.

Ich bisher erreicht :

1. val test = sqlContext.load(
        "com.databricks.spark.csv",
        Map("path" -> "test_set.csv", "header" -> "true", "inferSchema" -> "true", "delimiter" -> "," , "quote" -> "\"", "escape" -> "\\" ,"parserLib" -> "univocity" ))#read a csv file

 2.   val dataframe = test.toDF() #convert to dataframe

  3.    dataframe.foreach(println) #print

    4. dataframe.map(row => {
        val row4 = row.getAs[String](4)
        val make = row4.replaceAll("[\r\n]", "|") 
        (make)
      }).collect().foreach(println) #replace not working for me

Probe-set :

(17 , D73 ,525, 1  ,testing\n    ,  90 ,20.07.2011 ,null ,F10 , R)
 (17 , D73 ,526, 1  ,null         ,  89 ,20.07.2011 ,null ,F10 , R)
 (17 , D73 ,529, 1  ,once \n again,  10 ,20.07.2011 ,null ,F10 , R)
 (17 , D73 ,531, 1  ,test3\n      ,  10 ,20.07.2011 ,null ,F10 , R)

Erwartete Ergebnis festgelegt :

(17 , D73 ,525, 1  ,testing|    ,  90 ,20.07.2011 ,null ,F10 , R)
 (17 , D73 ,526, 1  ,null         ,  89 ,20.07.2011 ,null ,F10 , R)
 (17 , D73 ,529, 1  ,once | again,  10 ,20.07.2011 ,null ,F10 , R)
 (17 , D73 ,531, 1  ,test3|      ,  10 ,20.07.2011 ,null ,F10 , R)

was für mich gearbeitet:

val rep = "\n123\n Main Street\n".replaceAll("[\\r\\n]", "|") rep: String = |123| Main Street|

aber warum bin ich nicht in der Lage zu tun, auf Tupel-basis?

 val dataRDD = lines_wo_header.map(line => line.split(";")).map(row => (row(0).toLong, row(1).toString, 
                                               row(2).toLong, row(3).toLong, 
                                               row(4).toString, row(5).toLong,
                                               row(6).toString, row(7).toString, row(8).toString,row(9).toString)) 

dataRDD.map(row => {
                val wert = row._5.replaceAll("[\\r\\n]", "|") 
                (row._1,row._2,row._3,row._4,wert,row._6, row._7,row._8,row._9,row._10)
                }).collect().foreach(println)

Funke --version 1.3.1

InformationsquelleAutor user3560220 | 2016-05-02

2

Wenn du Spark SQL 1.5 oder höher, können Sie erwägen, die Funktionen verfügbar für Spalten. Vorausgesetzt, Sie wissen nicht (oder nicht haben) die Namen für die Spalten, die Sie tun können, wie im folgenden Codeausschnitt gezeigt:
```
val df = test.toDF()

import org.apache.spark.sql.functions._
val newDF = df.withColumn(df.columns(4), regexp_replace(col(df.columns(4)), "[\\r\\n]", "|"))
```
Wenn Sie wissen, den Namen der Spalte, die Sie ersetzen können df.columns(4) durch seinen Namen die in beiden vorkommen.

Ich hoffe, das hilft.
Cheers.
- thx, ich habe leider keine DataFrame API. Arbeiten müssen, um es mit RDDs.
- Sie können die Arbeit mit den dataframe und konvertieren es zurück zu RDD später durch aufrufen newDF.rdd oder newDF.map
- Ich habe versucht, Ihre Lösung, aber das "\n" - Zeichen wurden nicht ersetzt. Wie könnte man es lösen. ? Alles, was ich bin fehlt ?
- Eigentlich, in meinen tests, auch deine Lösung funktioniert. Vielleicht ist es etwas, das in Ihren Daten? Ist es einen Fehler auslösen, oder es funktioniert einfach nicht ersetzen?
- Ich habe versucht die Lösung wie von Ihnen vorgeschlagen und versucht, um meine Ausgabe-Datei lokal mit standalone-Funke in windows. Wenn ich öffnen Sie in Notepad++, es zeigt immer CRLF am Ende jeder Zeile mit den Einstellungen --> Ansicht-symbol aktiviert. Ich erwartete, dass ich die CRLF Weg zu gehen, mit der ersetzen-Funktion. Bin ich hier etwas fehlt? regexp_replace(ColName, "[\\r\\n]", "") as trimmed_column
InformationsquelleAutor Daniel de Paula

Meine Idee war, ersetzen Sie die \n-Zeichen mit der pipe "|" in der Funke.

Versuchte ich replaceAll Methode, aber es funktioniert nicht. Hier ist eine alternative, um das gleiche zu erreichen:

val test = sq.load(
        "com.databricks.spark.csv",
        Map("path" -> "file:///home/veda/sample.csv", "header" -> "false", "inferSchema" -> "true", "delimiter" -> "," , "quote" -> "\"", "escape" -> "\\" ,"parserLib" -> "univocity" ))

val dataframe = test.toDF()

val mapped = dataframe.map({
    row => {
    val str = row.get(0).toString()
    var fnal=new StringBuilder(str)
    //replace newLine 
    var newLineIndex=fnal.indexOf("\\n")
    while(newLineIndex != -1){
        fnal.replace(newLineIndex,newLineIndex+2,"|")
        newLineIndex = fnal.indexOf("\\n")                  
    }

    //replace carriage returns
    var cgIndex=fnal.indexOf("\\r")
    while(cgIndex != -1){
        fnal.replace(cgIndex,cgIndex+2,"|")
        cgIndex = fnal.indexOf("\\r")                   
    }

    (fnal.toString()) //tuple modified

    }
})

mapped.collect().foreach(println)

Hinweis: möchten Sie möglicherweise verschieben Sie die doppelten code zu trennen-Funktion.

InformationsquelleAutor mrnakumar

Multi-line-support für CSV Hinzugefügt, spark version 2.2 JIRA und spark-2.2 ist noch nicht freigegeben.

Hatte ich konfrontiert gleiche Problem und gelöst mit die uns helfen, hadoop-input-format und reader.

Kopieren InputFormat-und reader-Klassen aus git und umzusetzen, wie diese:

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;

//implementation

 JavaPairRDD<LongWritable, Text> rdd =
                context.
                        newAPIHadoopFile(path, FileCleaningInputFormat.class, null, null, new Configuration());
JavaRDD<String> inputWithMultiline= rdd.map(s -> s._2().toString())

Eine andere Lösung- Nutzung CSVInputFormat von Apache crunch zu Lesen CSV-Datei dann analysieren jede CSV-Zeile mit opencsv:

sparkContext.newAPIHadoopFile(path, CSVInputFormat.class, null, null, new Configuration()).map(s -> s._2().toString());

Apache crunch maven-dependency:

 <dependency>
      <groupId>org.apache.crunch</groupId>
      <artifactId>crunch-core</artifactId>
      <version>0.15.0</version>
  </dependency>

InformationsquelleAutor Rahul Sharma

Schreibe einen Kommentar

Du musst angemeldet sein, um einen Kommentar abzugeben.