2011-10-10 83 views
0

下面是这种情况Hadoop的流:写输出到不同的文件

  Reducer1 
     / 
Mapper - - Reducer2 
     \ 
      ReducerN 

在减速,我想写在不同文件中的数据,可以说减速看起来像

def reduce(): 
    for line in sys.STDIN: 
    if(line == type1): 
     create_type_1_file(line) 
    if(line == type2): 
     create_type_2_file(line) 
    if(line == type3): 
     create_type3_file(line) 
     ... and so on 
def create_type_1_file(line): 
    # writes to file1 
def create_type2_file(line): 
    # writes to file2 
def create_type_3_file(line): 
    # write to file 3 

考虑路径写为:

file1 = /home/user/data/file1 
file2 = /home/user/data/file2 
file3 = /home/user/data/file3 

当我在pseudo-distributed mode(machine with one node and hdfs daemons running)运行,事情是因为所有的d好aemons将写入同一组文件

问题: - 如果我在1000台机器的群集中运行此操作,它们是否会写入同一组文件?我在这种情况下是writing to local filesystem
- 在hadoop streaming有没有更好的方法来执行此操作?

谢谢

+0

这个答案可能会帮助(不知道的,因此评论

Job job = new Job(); FileInputFormat.setInputPath(job, inDir); //outDir is the root path, in this case, outDir="/home/user/data/" FileOutputFormat.setOutputPath(job, outDir); //You have to assign the output formatclass.Using MultipleOutputs in this way will still create zero-sized default output, eg part-00000. To prevent this use LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class); instead of job.setOutputFormatClass(TextOutputFormat.class); in your Hadoop job configuration. LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(Text.class); job.setMapperClass(MOMap.class); job.setReducerClass(MOReduce.class); ... job.waitForCompletion(true); 

在减速用法)http://stackoverflow.com/questions/162 6786 /生成-分离 - 输出 - 文件功能于Hadoop的流/ 1690092#1690092 – Nija

回答

0

通常减少的O/P被写入像HDFS一个可靠的存储系统,因为如果其中一个节点出现故障则与该节点丢失有关的减少数据。在Hadoop框架的上下文之外再次运行特定的reduce任务是不可能的。另外,一旦作业完成,来自1000个节点的o/p必须针对不同的输入类型进行合并。

HDFS中的并行写入是not supported。可能存在多个缩减器可能正在写入HDFS中的同一文件的情况,这可能会破坏文件。当多个reduce任务在单个节点上运行时,同时写入单个本地文件时可能会出现并发问题。

其中一个解决方案是有reduce task specific file name,后来将所有文件合并为一个特定的输入类型。

0

可以使用MultipleOutputs类从Reducer将输出写入多个位置。您可以将file1,file2和file3视为三个文件夹,并分别向这些文件夹写入1000个Reducers的输出数据。作业提交


使用模式:

private MultipleOutputs out; 

public void setup(Context context) { 

    out = new MultipleOutputs(context); 

    ... 

} 

public void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { 

//'/' characters in baseOutputPath will be translated into directory levels in your file system. Also, append your custom-generated path with "part" or similar, otherwise your output will be -00000, -00001 etc. No call to context.write() is necessary. 
for (Text line : values) { 

    if(line == type1) 
     out.write(key, new Text(line),"file1/part"); 

    else if(line == type2) 
     out.write(key, new Text(line),"file2/part"); 

else if(line == type3) 
     out.write(key, new Text(line),"file3/part"); 
    } 
} 

protected void cleanup(Context context) throws IOException, InterruptedException { 
     out.close(); 
    } 

裁判:https://hadoop.apache.org/docs/r2.6.3/api/org/apache/hadoop/mapreduce/lib/output/MultipleOutputs.html