Hadoop TextOutputFormat：将标头添加到 CSV 输出答案

【问题标题】：Hadoop TextOutputFormat: add headers to CSV outputHadoop TextOutputFormat：将标头添加到 CSV 输出
【发布时间】：2013-08-24 08:12:50
【问题描述】：

我正在维护一个简单的 hadoop 作业，它生成 CSV 文件作为 HDFS 中的输出。该作业使用 TextOutputFormat。我想将前导标题行添加到 csv 文件（我知道部分文件是由不同的工作人员创建的，如果他们每个人都有标题，那不是问题）。如何做到这一点？

编辑：级联可以help 但乍一看我不想开始使用新框架

编辑：

所以我想为输出的 CSV 文件添加标题。列数是确定性的。这是我的 Reducer 类的骨架：

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;

public final class Reducer extends Reducer<Text, IntWritable, Text, IntWritable>
{
    private MultipleOutputs<Text, IntWritable> mos;

    private static final Text KEY_HOLDER = new Text();

    private static final IntWritable VALUE_HOLDER = new IntWritable(1);

    @Override
    public void setup(final Context context)
    {
        mos = new MultipleOutputs<Text, IntWritable>(context);
    }

    @Override
    public void cleanup(final Context context) throws IOException, InterruptedException
    {
        mos.close();
    }

    @Override
    public void reduce(final Text key, final Iterable<IntWritable> values, final Context context)
            throws IOException, InterruptedException
    {
        // [... some business logic ...]        
        mos.write(KEY_HOLDER, VALUE_HOLDER, "myFileName");
        context.progress();
    }
}

【问题讨论】：

为什么投反对票？如果这个问题没有意义，请告诉我。
你能在你的mapper或reducer中添加header吗？您可以在实际数据之前输出表头。
请显示一些代码，根据您的描述，我们无法帮助您。
@zsxwing 映射器接收一行，reducer 接收一个元组和一些数值......它们如何在“好吧这是第一行，添加标题.. ." ?
可以在setup方法中输出header。

标签： java csv hadoop mapreduce

【解决方案1】：

您可以覆盖 mapper/reducer 类中的 run() 以根据您的要求添加标头。例如。如果你想在你的 final o/p 中添加 FisrtName 和 LastName。你可以使用下面的代码作为参考。

public void run(Context context) throws IOException, InterruptedException
  {
        setup(context);
        column = new Text("ColumnName") ;
        values = new Text("FirstName" + "\t" + "LastName") ;
        context.write(column, values);
        try
        {
          while (context.nextKey())
          {
            reduce(context.getCurrentKey(), context.getValues(), context);
            Iterator<IntWritable> iter = context.getValues().iterator();
            if(iter instanceof ReduceContext.ValueIterator)
            {              ((ReduceContext.ValueIterator<IntWritable>)iter).resetBackupStore();        
            }
          }
        }
        finally
        {
          cleanup(context);
        }
  }

【讨论】：