在 hdfs 上将 csv 文件转换为另一种 csv 格式答案

【问题标题】：Convert csv files to another csv format on hdfs在 hdfs 上将 csv 文件转换为另一种 csv 格式
【发布时间】：2015-02-02 11:12:25
【问题描述】：

我必须实现一个 CSV 文件转换器才能在 hadoop 集群上运行。主要线路是：

我在 hdfs 上有一堆 csv 文件，内容随意。
我知道如何使用 java 代码将它们转换为“标准”（即具有指定行）。
转换需要一些参数（大约 10 或 15 个），每个文件都不同。
我不介意要分段的输出文件。
但我希望它们有一个 input-filename[##].csv 名称来区分它们以供以后处理/可视化。

我的问题是：最好的方法是什么？

作为 hadoop 的新手，我正在考虑使用 map reduce 来执行此操作，但我对输出格式有疑问。另一方面，我可以使用 spark（在 scala 中使用我的 java 代码）。编码似乎很容易，但我不知道该怎么做。

非常感谢（更多）有经验的用户对要实施的主要任务提出意见。

【问题讨论】：

您的问题有点含糊，因为这在很大程度上取决于您的集群当前安装了哪些工具、您的 CSV 文件的大小、您是经常运行这项工作还是只运行一次。当然 MapReduce 在这里并不是很有用，因为你没有任何东西可以键入。您可以编写一个独立的映射器类。然而，这似乎过于复杂。根据您所说的，我会选择 Pig，因为它是轻量级的，并且可能是最快的编码。为这么简单的事情编写大量 Java 是没有意义的。
你说得对，我说的太含糊了，但这是因为我不知道哪些信息是相关的。我把 pig 放在一边，因为转换不是那么简单，而且我已经有了它的代码。
啊，如果你已经有代码并且可以访问 Spark，那就去吧。

标签： java csv hadoop mapreduce

【解决方案1】：

Spark 是一个不错的选择。它还通过快速处理为您提供更大的灵活性。

【讨论】：

【解决方案2】：

用 spark 确实很简单：

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;

import org.apache.hadoop.fs.FileUtil;

import java.io.File;

public class Converter {
    static String appName = "CSV-Conversion";  // spark app name
    static String master = "local";            // spark master 

    JavaSparkContext sc;

    /**
     * Init spark context
     */
    public Converter(){
        SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);
        sc = new JavaSparkContext(conf);
    }

    /**
     * The conversion using spark
     */
    public void convertFile(String inputFile, String outputDir){
        JavaRDD<String> inputRdd = sc.textFile(inputFile);
        JavaRDD<String> outputRdd = inputRdd.map(Converter::convertLine);
        outputRdd.saveAsTextFile(outputDir);
    }

    /**
     * The function that convert each file line.
     *
     * It is static (i.e. does not requires 'this') and does not use other object.
     * If external objects (or not static method) are required, they must be
     * serializable so that a copy can be send to each worker node.
     * It is however better to avoid or at least minimize such data transfer.
     */
    public static String convertLine(String line){
        return line.toUpperCase();
    }


    /**
     * As a stand-alone app
     */
    public static void main(String[] args){
        if(args.length!=2) {
            System.out.println("Invalid number of arguments. Usage: Converter inputFile outputDir");
            System.exit(1);
        }

        String inputFile = args[0];
        String outputDir = args[1];

        FileUtil.fullyDelete(new File(outputDir));

        Converter c = new Converter();
        c.convertFile(inputFile,outputDir);
    }
}

我发了simple maven project for it in github

【讨论】：