Jak wyeksportować dane z Spark SQL do CSV

Od Spark 2.X spark-csv jest zintegrowany jako natywne źródło danych . Dlatego konieczne polecenie upraszcza do (windows)

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

Lub UNIX

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

Powyższa odpowiedź w spark-csv jest poprawna, ale jest problem-biblioteka tworzy kilka plików w oparciu o partycjonowanie ramki danych. A to nie jest to, czego zwykle potrzebujemy. Możesz więc połączyć wszystkie partycje w jedną:

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

I zmienić nazwę wyjścia lib (nazwa "part-00000") na pożądaną nazwę pliku.

Ten wpis na blogu zawiera więcej szczegółów: https://fullstackml.com/2015/12/21/how-to-export-data-frame-from-apache-spark/

Najprostszym sposobem jest mapowanie na RDD ramki danych i użycie mkstringu:

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

Od Spark 1.5 (a nawet wcześniej) df.map(r=>r.mkString(",")) zrobiłby to samo jeśli chcesz ucieczki CSV możesz użyć Apache commons lang do tego. np. oto kod, którego używamy

 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)
  }

Komunikat o błędzie sugeruje, że nie jest to obsługiwana funkcja w języku zapytań. Ale możesz zapisać ramkę danych w dowolnym formacie, jak zwykle poprzez interfejs RDD (df.rdd.saveAsTextFile). Lub możesz sprawdzić https://github.com/databricks/spark-csv .

Za pomocą spark-csv możemy zapisać plik CSV.

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