hadoop - 如何使用Cascading的CoGroup

Question

我的递归目录结构具有不同数量的零件文件。我想对这些文件应用 CoGroup。

假设，我的目录结构是这样的:

directory1/dir1/part-0000
               /part-0001
               /part-0002
           dir2/part-0000
               /part-0001
               /part-0002
           dir3/part-0000
               /part-0001
               /part-0002
           dir4/part-0000
               /part-0001
               /part-0002

这些零件文件包含制表符分隔的数据，例如:
field1 field2 field3 field4 field5

我想合并所有具有公共(public)值 field1、field3、field4 和 field5 的部分文件。也就是说，最终输出文件将包含如下数据:

field1 field2_dir1_files field2_dir2_files field2_dir3_files field2_dir4_files field3 field4 field5

如果有任何 MapReduce 解决方案，非常欢迎您，我也会尝试 :)
使用 Cascading CoGroup API 怎么可能？ 请帮我解决这个问题，我一直在努力解决过去两周的这个问题。

提前致谢!

Answer 1

这里我们可以通过使用级联中提供的简单混合连接来解决这个问题

http://docs.cascading.org/cascading/2.5/javadoc/cascading/pipe/joiner/MixedJoin.html

首先将每个输入路径连接到每个管道，并合并与目录相关的管道。

让合并输出管道我们得到 dir1,dir2,dir3 将有文件

字段1字段2字段3字段4字段5

并创建这些管道的数组作为 dir[]

为每个管道创建我们要连接的每个管道的连接字段数组 字段1 字段3 字段4 字段5

Fields outputFields =new Fields("field1","field2_dir1_files","field3","field4","field5","2field1","2field2_dir2_files","2field3","2field4","2field5","3field1","field2_dir3_files","3field3","3field4","3field5");

boolean[] i ={false,false,false}

Pipe LastJoin = joiningPipe = new CoGroup(dir[],JoinFields[],new MixedJoin(outputFields,i);

Pipe required = new Retain("field1","field2_dir1_files","field2_dir2_files","field2_dir3_files","field3","field4","field5");

retain inode 保留输出中需要的字段