Ersetzen von Escape-Zeichen in Funke

Question

Der Versuch, alle \ Zeichen in einer CSV zu ersetzen, damit sie korrekt gelesen werden. Hier ist meine UDF:

def escapeBackslash: String => String = _.replaceAll("[_\\]","\\\\") 
def escapeBackslashUDF = udf(escapeBackslash) 

\ N ist in Ordnung, so brauche ich nicht um diejenigen zu kümmern, aber ich bin immer die Ausgabe:

123,myName\,myDesc,245,true 
345,anotherName,\N,600,\N 
789,name3,desc3,\N,false 

Jede Hilfe da draußen?

Hier ist der Eingang:

val schema = StructType(Seq(StructField("id", StringType), StructField("name", StringType), StructField("dec", StringType), StructField("amount",IntegerType), StructField("enabled",BooleanType))) 
val rdd = spark.sparkContext.parallelize(Seq(Row("123", "myName\\", "myDesc",245, true), Row("345","anotherName",null,600,null), Row("789","name3","desc3",null,false))) 

Ich versuche, die folgende Ausgabe erhalten die zu entkommen:

123,myName\\,myDesc,245,true 
345,anotherName,\N,600,\N 
789,name3,desc3,\N,false 

Answer 1

Was ich dies für jeden String-Feld ein UDF schreiben war zu beheben hat, registrierte die UDF, sortierte dann durch jedes Feld, um Strings zu finden, und wählte diese Felder mit der UDF aus.

val func:(String) => String = (test:String) => { 
    Option(test) match{ 
    case Some(t) => t.replace("\\", "\\\\") 
    case _ => <null> 
    } 

}

val fields = dfOut.schema.fields 
    .map(col => { 
    if (col.dataType.isInstanceOf[StringType]){ s"${EscapeBackslashUDF.name}(${col.name})"} 
    else { s"${col.name}" } 
    }).toSeq 

super.write(target, dfOut.selectExpr(fields:_*))