Wie

Question

arbeitet mit HiveQL Dieses Befehl in der CSV-Daten von Spark-SQL exportieren:

java.lang.RuntimeException: Unsupported language features in query: 
    insert overwrite directory '/data/home.csv' select * from testtable 

Bitte leiten: Ich erhalte eine Fehlermeldung mit einem org.apache.spark.sql.hive.HiveQl Stack-Trace

insert overwrite directory '/data/home.csv' select * from testtable; 

Aber mit Spark-SQL Ich schreibe Export in CSV-Funktion in Spark SQL.

Answer 1

Sie unter Anweisung verwenden können, um den Inhalt von Datenrahmen im CSV-Format df.write.csv("/data/home/csv")

zu schreiben Wenn Sie den gesamten Datenrahmen in eine einzige CSV-Datei schreiben müssen, verwenden Sie dann df.coalesce(1).write.csv("/data/home/sample.csv")

Für Funken 1 .x, können Sie spark-csv verwenden, um die Ergebnisse in CSV-Dateien zu schreiben

Below scala Schnipsel hel würde p

import org.apache.spark.sql.hive.HiveContext 
// sc - existing spark context 
val sqlContext = new HiveContext(sc) 
val df = sqlContext.sql("SELECT * FROM testtable") 
df.write.format("com.databricks.spark.csv").save("/data/home/csv") 

Um den Inhalt in eine einzige Datei

import org.apache.spark.sql.hive.HiveContext 
// sc - existing spark context 
val sqlContext = new HiveContext(sc) 
val df = sqlContext.sql("SELECT * FROM testtable") 
df.coalesce(1).write.format("com.databricks.spark.csv").save("/data/home/sample.csv") 

Answer 2

Die Fehlermeldung schlägt vor, dass dies keine unterstützte Funktion in der Abfragesprache ist. Sie können einen DataFrame jedoch wie gewohnt über die RDD-Schnittstelle (df.rdd.saveAsTextFile) in einem beliebigen Format speichern. Oder Sie können https://github.com/databricks/spark-csv auschecken.

Answer 3

Der einfachste Weg ist, zu schreiben über den Datenrahmen des RDD zur Karte und verwendet mkString:

df.rdd.map(x=>x.mkString(",")) 

Als Spark 1.5 (oder schon vorher) df.map(r=>r.mkString(",")) würde das gleiche tun Wenn Sie CSV-Escaping möchten, können Sie Apache Commons lang dafür verwenden. z.B. hier ist der Code, den wir

def DfToTextFile(path: String, 
        df: DataFrame, 
        delimiter: String = ",", 
        csvEscape: Boolean = true, 
        partitions: Int = 1, 
        compress: Boolean = true, 
        header: Option[String] = None, 
        maxColumnLength: Option[Int] = None) = { 

    def trimColumnLength(c: String) = { 
     val col = maxColumnLength match { 
     case None => c 
     case Some(len: Int) => c.take(len) 
     } 
     if (csvEscape) StringEscapeUtils.escapeCsv(col) else col 
    } 
    def rowToString(r: Row) = { 
     val st = r.mkString("~-~").replaceAll("[\\p{C}|\\uFFFD]", "") //remove control characters 
     st.split("~-~").map(trimColumnLength).mkString(delimiter) 
    } 

    def addHeader(r: RDD[String]) = { 
     val rdd = for (h <- header; 
        if partitions == 1; //headers only supported for single partitions 
        tmpRdd = sc.parallelize(Array(h))) yield tmpRdd.union(r).coalesce(1) 
     rdd.getOrElse(r) 
    } 

    val rdd = df.map(rowToString).repartition(partitions) 
    val headerRdd = addHeader(rdd) 

    if (compress) 
     headerRdd.saveAsTextFile(path, classOf[GzipCodec]) 
    else 
     headerRdd.saveAsTextFile(path) 
    } 

Answer 4

Die Antwort oben mit Funken csv korrekt verwenden, aber es gibt ein Problem - die Bibliothek mehrere Dateien auf der Grundlage der Datenrahmenpartitionierung erstellt. Und das brauchen wir normalerweise nicht. So können Sie alle Partitionen zu einem kombinieren:

df.coalesce(1). 
    write. 
    format("com.databricks.spark.csv"). 
    option("header", "true"). 
    save("myfile.csv") 

und benennen Sie die Ausgabe des lib (Namen „Teil-00000“) zu einem Wunsch Dateinamen.

Dieser Blog-Eintrag liefert weitere Details: https://fullstackml.com/2015/12/21/how-to-export-data-frame-from-apache-spark/

Answer 5

Da Funken 2.Xspark-csv als native datasource integriert ist. Daher vereinfacht die notwendige Anweisung (Fenster)

df.write 
    .option("header", "true") 
    .csv("file:///C:/out.csv") 

oder UNIX

df.write 
    .option("header", "true") 
    .csv("/var/out.csv") 

Answer 6

Mit Hilfe von Funken csv wir in eine CSV-Datei schreiben kann.

val dfsql = sqlContext.sql("select * from tablename") 
dfsql.write.format("com.databricks.spark.csv").option("header","true").save("output.csv")`