Hadoop Spark 1.4.1 - 对多个 CSV 文件进行排序并将排序结果保存在 1 个输出文件中

Question

我在 HDFS 中有 3 个文件，我想使用最有效的方法先在第一列排序，然后在第二列排序，然后在 Spark 1.4 中使用 Scala（或 Python）将排序结果存储回 HDFS 上的新文件。 1：
hdfs:///test/2016/file.csv
hdfs:///test/2015/file.csv
hdfs:///test/2014/file.csv

文件看起来像这样（没有标题）：
hdfs:///test/2016/file.csv
127,56,abc
125,56,abc
121,56,abc

hdfs:///test/2016/file.csv
126,66,abc
122,56,abc
123,46,abc

hdfs:///test/2016/file.csv
122,66,abc
128,56,abc
123,16,abc

排序后的输出要保存到 HDFS：
hdfs:///test/output/file.csv
121,56,abc
122,56,abc
122,66,abc
123,16,abc
123,46,abc
125,56,abc
126,66,abc
127,56,abc
128,56,abc

我对 Spark 很陌生，到目前为止我只知道如何加载文件：
val textFile = sc.textFile("hdfs:///test/2016/file.csv")

试图在互联网上阅读如何排序，但不清楚哪些库应该适用于这种情况（CSV 文件）和这个版本的 Spark（1.4.1）以及如何使用它们。 请帮忙，乔

Answer 1

我建议使用 databricks csv 库来读写 csv：https://github.com/databricks/spark-csv

由于我现在无法访问 hdfs，因此此示例使用文件系统，但在与 hdfs 路径一起使用时也应该可以工作。

import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkContext, SparkConf}
import org.apache.spark.sql.functions._  // needed for ordering the dataframe

object StackoverflowTest {

  def main(args: Array[String]) {
    // basic spark init
    val conf = new SparkConf().setAppName("Data Import from CSV")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    // first we load every file from the data directory that starts with 'file'
    val storeDf = sqlContext.read
      .format("com.databricks.spark.csv")
      .option("inferSchema", "true")
      .load("data/file*")

    // then we sort it and write to an output
    storeDf
      .orderBy("C0", "C1")  // default column names
      .repartition(1)   // in order to have 1 output file
      .write
      .format("com.databricks.spark.csv")
      .save("data/output")
  }

}

结果将作为 csv 写入 data/output/part-00000。 希望这会有所帮助。

Answer 2

val textFile = sc.textFile("hdfs:///test/*/*.csv")
                 .map( _.split(",",-1) match { case Array(col1, col2, col3) => (col1, col2, col3) })
                 .sortBy(_._1)
                 .map(_._1+","+_._2+","+_._3)
                 .saveAsTextFile("hdfs:///testoutput/output/file.csv")

您需要保存在不同的文件夹中，否则您生成的文件将在您再次运行时被重复使用。