Как экспортировать данные из искрового SQL в CSV

Question

Эта команда работает с HiveQL:

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

Но Спарк SQL я получаю сообщение об ошибке с трассы org.apache.spark.sql.hive.HiveQl стека:

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

Пожалуйста, руководство мне написать экспорт в CSV-функцию в Spark SQL.

Answer 1

Вы можете использовать ниже заявление, чтобы записать содержимое dataframe в формате CSV df.write.csv("/data/home/csv")

Если вам нужно записать весь dataframe в один файл CSV, а затем использовать df.coalesce(1).write.csv("/data/home/sample.csv")

Для искры 1 .x, вы можете использовать spark-csv для записи результатов в CSV-файлы

Ниже Скале сниппет будет HEL р

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

Чтобы записать содержимое в один файл

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

Сообщение об ошибке предполагает, что это не поддерживается в языке запросов. Но вы можете сохранить DataFrame в любом формате, как обычно, через интерфейс RDD (df.rdd.saveAsTextFile). Или вы можете проверить https://github.com/databricks/spark-csv.

Answer 3

Самый простой способ отобразить более РДУ в DataFrame и использовать mkString:

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

На Спарк 1.5 (или даже до этого) df.map(r=>r.mkString(",")) будет делать то же самое , если вы хотите, чтобы сбрасывание CSV вы можете использовать для этого apache commons lang. например вот код, который мы используем

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

Ответ на этот вопрос выше искровым CSV является правильным, но есть проблема - библиотека создает несколько файлов на основе кадра данных разделов. И это не то, что нам обычно нужно. Таким образом, вы можете объединить все разделы к одному:

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

и переименовать вывод Lib (название «часть-00000») на желание файла.

Этот блог содержит более подробную информацию: https://fullstackml.com/2015/12/21/how-to-export-data-frame-from-apache-spark/

Answer 5

С искрой 2.Xspark-csv интегрирован в native datasource. Таким образом, необходимо утверждение упрощается (окон)

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

или UNIX

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

Answer 6

С помощью искрового CSV можно записать в файл CSV.

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