知名室内设计网站,廊坊快速排名优化,昨天正常的网站突然显示建设中,尤溪网站建设在本文中#xff0c;我们继续执行一系列实现算法的系列#xff0c;该算法在使用MapReduce进行数据密集型文本处理中找到#xff0c;这一次讨论数据联接。 虽然我们将讨论在Hadoop中联接数据的技术并提供示例代码#xff0c;但在大多数情况下#xff0c;您可能不会自己编写… 在本文中我们继续执行一系列实现算法的系列该算法在使用MapReduce进行数据密集型文本处理中找到这一次讨论数据联接。 虽然我们将讨论在Hadoop中联接数据的技术并提供示例代码但在大多数情况下您可能不会自己编写代码来执行联接。 取而代之的是使用可以在更高抽象级别工作的工具例如Hive或Pig可以更好地完成连接数据。 如果有可以帮助您处理数据的工具为什么还要花时间学习如何联接数据呢 可以说联接数据是Hadoop的最大用途之一。 全面了解Hadoop如何执行联接对于确定使用哪个联接以及在出现问题时进行调试至关重要。 此外一旦您完全了解了Hadoop中如何执行不同的联接就可以更好地利用Hive和Pig等工具。 最后在一种情况下一种工具可能无法满足您的需求因此您必须袖手旁观并自行编写代码。 加入的需要 在处理大型数据集时如果不是必需的话通过公用密钥连接数据的需求可能会非常有用。 通过加入数据您可以进一步获得洞察力例如加入时间戳以将事件与一天中的时间关联起来。 连接数据的需求多种多样。 我们将在3个单独的帖子中介绍3种类型的联接Reduce-Side联接Map-Side联接和Memory-Backed联接。 在这一期中我们将考虑使用Reduce-Side联接。 减少侧面连接 在我们将要讨论的联接模式中减少端联接是最容易实现的。 简化方联接的直接原因是Hadoop将相同的密钥发送到相同的reducer因此默认情况下数据是为我们组织的。 要执行联接我们只需要缓存一个密钥并将其与传入密钥进行比较。 只要键匹配我们就可以结合来自相应键的值。 由于所有数据在整个网络上都经过混洗因此使用减少侧连接进行权衡是性能。 在减少侧连接中我们将考虑两种不同的方案一对一和一对多。 我们还将探索不需要跟踪传入密钥的选项 给定键的所有值都将在简化器中分组在一起。 一对一加入 一对一联接的情况是数据集“ X”中的值与数据集“ Y”中的值共享一个公共密钥。 由于Hadoop保证将相等的键发送到同一reducer因此在两个数据集上进行映射将为我们处理联接。 由于仅对键进行排序因此值的顺序未知。 我们可以使用辅助排序轻松解决这种情况。 我们二级排序的实现方式是用“ 1”或“ 2”标记键以确定值的顺序。 我们需要采取一些额外的步骤来实施我们的标记策略。 实现一个WritableComparable 首先我们需要编写一个实现WritableComparable接口的类该接口将用于包装密钥。 public class TaggedKey implements Writable, WritableComparableTaggedKey {private Text joinKey new Text();private IntWritable tag new IntWritable();Overridepublic int compareTo(TaggedKey taggedKey) {int compareValue this.joinKey.compareTo(taggedKey.getJoinKey());if(compareValue 0 ){compareValue this.tag.compareTo(taggedKey.getTag());}return compareValue;}//Details left out for clarity} 当我们对TaggedKey类进行排序时具有相同joinKey值的键将在tag字段的值上进行次要排序以确保我们想要的顺序。 编写自定义分区程序 接下来我们需要编写一个自定义分区程序该分区程序仅在确定复合键和数据发送到哪个减速器时才考虑连接键 public class TaggedJoiningPartitioner extends PartitionerTaggedKey,Text {Overridepublic int getPartition(TaggedKey taggedKey, Text text, int numPartitions) {return taggedKey.getJoinKey().hashCode() % numPartitions;}
} 至此我们拥有了连接数据并确保值顺序的条件。 但是当键进入reduce()方法时我们不想跟踪它们。 我们希望将所有价值观归为一体。 为此我们将使用Comparator 该Comparator在决定如何对值进行分组时仅考虑联接键。 编写组比较器 用于分组的比较器如下所示 public class TaggedJoiningGroupingComparator extends WritableComparator {public TaggedJoiningGroupingComparator() {super(TaggedKey.class,true);}Overridepublic int compare(WritableComparable a, WritableComparable b) {TaggedKey taggedKey1 (TaggedKey)a;TaggedKey taggedKey2 (TaggedKey)b;return taggedKey1.getJoinKey().compareTo(taggedKey2.getJoinKey());}
}数据结构 现在我们需要确定将用于密钥的哪些数据。 对于我们的样本数据我们将使用从Fakenames Generator生成的CSV文件。 第一列是GUID它将用作我们的联接键。 我们的样本数据包含诸如姓名地址电子邮件工作信息信用卡和拥有的汽车之类的信息。 为了演示的目的我们将使用GUID名称和地址字段并将它们放置在一个结构如下的文件中 cdd8dde3-0349-4f0d-b97a-7ae84b687f9c,Esther,Garner,4071 Haven Lane,Okemos,MI
81a43486-07e1-4b92-b92b-03d0caa87b5f,Timothy,Duncan,753 Stadium Drive,Taunton,MA
aef52cf1-f565-4124-bf18-47acdac47a0e,Brett,Ramsey,4985 Shinn Street,New York,NY 然后我们将使用GUID电子邮件地址用户名密码和信用卡号字段然后将其放置在另一个文件中该文件应类似于 cdd8dde3-0349-4f0d-b97a-7ae84b687f9c,517-706-9565,EstherJGarnerteleworm.us,Waskepter38,noL2ieghie,MasterCard,
5305687295670850
81a43486-07e1-4b92-b92b-03d0caa87b5f,508-307-3433,TimothyDDuncaneinrot.com,Conerse,Gif4Edeiba,MasterCard,
5265896533330445
aef52cf1-f565-4124-bf18-47acdac47a0e,212-780-4015,BrettMRamseydayrep.com,Subjecall,AiKoiweihi6,MasterCard,524 现在我们需要有一个Mapper它将知道如何处理我们的数据以提取正确的联接键并设置正确的标签。 创建映射器 这是我们的Mapper代码 public class JoiningMapper extends MapperLongWritable, Text, TaggedKey, Text {private int keyIndex;private Splitter splitter;private Joiner joiner;private TaggedKey taggedKey new TaggedKey();private Text data new Text();private int joinOrder;Overrideprotected void setup(Context context) throws IOException, InterruptedException {keyIndex Integer.parseInt(context.getConfiguration().get(keyIndex));String separator context.getConfiguration().get(separator);splitter Splitter.on(separator).trimResults();joiner Joiner.on(separator);FileSplit fileSplit (FileSplit)context.getInputSplit();joinOrder Integer.parseInt(context.getConfiguration().get(fileSplit.getPath().getName()));}Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {ListString values Lists.newArrayList(splitter.split(value.toString()));String joinKey values.remove(keyIndex);String valuesWithOutKey joiner.join(values);taggedKey.set(joinKey, joinOrder);data.set(valuesWithOutKey);context.write(taggedKey, data);}} 让我们回顾一下setup()方法中发生的事情。 首先从启动作业时在“配置”中设置的值获取连接键的索引和文本中使用的分隔符。 然后我们创建一个Guava拆分器用于拆分从对context.getConfiguration().get(separator)的调用中检索到的分隔符上的数据。 我们还创建了一个Guava Joiner用于在提取密钥后将数据重新放在一起。 接下来我们获取此映射器将要处理的文件的名称。 我们使用文件名来提取此配置中存储的文件的连接顺序。 我们还应该讨论map()方法中发生的事情 分散数据并创建值列表 从列表中删除联接密钥 重新将数据重新合并为单个字符串 设置连接密钥连接顺序和剩余数据 写出数据 因此我们已经读入数据提取了密钥设置了连接顺序然后将数据写回了。 让我们看一下如何结合数据。 联接数据 现在让我们看一下数据如何在化简器中联接 public class JoiningReducer extends ReduceTaggedKey, Text, NullWritable, Text {private Text joinedText new Text();private StringBuilder builder new StringBuilder();private NullWritable nullKey NullWritable.get();Overrideprotected void reduce(TaggedKey key, IterableText values, Context context) throws IOException, InterruptedException {builder.append(key.getJoinKey()).append(,);for (Text value : values) {builder.append(value.toString()).append(,);}builder.setLength(builder.length()-1);joinedText.set(builder.toString());context.write(nullKey, joinedText);builder.setLength(0);}
} 因为带有“ 1”标签的密钥首先到达了还原器所以我们知道名称和地址数据是第一个值而电子邮件用户名密码和信用卡数据是第二个值。 因此我们不需要跟踪任何键。 我们只需遍历值并将它们连接在一起。 一对一加入结果 这是运行我们的一对一MapReduce作业的结果 cdd8dde3-0349-4f0d-b97a-7ae84b687f9c,Esther,Garner,4071 Haven Lane,Okemos,MI,517-706-9565,EstherJGarnerteleworm.us,Waskepter38,noL2ieghie,MasterCard,
5305687295670850
81a43486-07e1-4b92-b92b-03d0caa87b5f,Timothy,Duncan,753 Stadium Drive,Taunton,MA,508-307-3433,TimothyDDuncaneinrot.com,Conerse,Gif4Edeiba,MasterCard,
5265896533330445
aef52cf1-f565-4124-bf18-47acdac47a0e,Brett,Ramsey,4985 Shinn Street,New York,NY,212-780-4015,BrettMRamseydayrep.com,Subjecall,AiKoiweihi6,MasterCard,
5243379373546690 正如我们可以看到的以上示例数据中的两条记录已合并为一条记录。 我们已经成功地将GUID名称地址电子邮件地址用户名密码和信用卡字段加入到一个文件中。 指定加入顺序 此时我们可能会问如何为多个文件指定连接顺序 答案就在我们的ReduceSideJoinDriver类中该类充当MapReduce程序的驱动程序。 public class ReduceSideJoinDriver {public static void main(String[] args) throws Exception {Splitter splitter Splitter.on(/);StringBuilder filePaths new StringBuilder();Configuration config new Configuration();config.set(keyIndex, 0);config.set(separator, ,);for(int i 0; i args.length - 1; i) {String fileName Iterables.getLast(splitter.split(args[i]));config.set(fileName, Integer.toString(i1));filePaths.append(args[i]).append(,);}filePaths.setLength(filePaths.length() - 1);Job job Job.getInstance(config, ReduceSideJoin);job.setJarByClass(ReduceSideJoinDriver.class);FileInputFormat.addInputPaths(job, filePaths.toString());FileOutputFormat.setOutputPath(job, new Path(args[args.length-1]));job.setMapperClass(JoiningMapper.class);job.setReducerClass(JoiningReducer.class);job.setPartitionerClass(TaggedJoiningPartitioner.class);job.setGroupingComparatorClass(TaggedJoiningGroupingComparator.class);job.setOutputKeyClass(TaggedKey.class);job.setOutputValueClass(Text.class);System.exit(job.waitForCompletion(true) ? 0 : 1);}
} 首先我们在第5行上创建一个番石榴分割器该分割器将用“ /”分割字符串。 然后在第8-10行上设置连接键的索引和文件中使用的分隔符。 在第12-17行中我们为要连接的输入文件设置标签。 命令行上文件名的顺序决定了它们在联接中的位置。 当我们从命令行循环遍历文件名时我们将拆分整个文件名并通过Guava Iterables.getLast()方法检索最后一个值基本文件名。 然后我们使用文件名作为键调用config.set() 并使用i 1作为值这将设置标签或连接顺序。 args数组中的最后一个值在循环中被跳过因为它用于第23行的MapReduce作业的输出路径。在循环的最后一行我们将每个文件路径附加到StringBuilder中稍后使用 22设置作业的输入路径。 我们只需要对所有文件使用一个映射器即JoiningMapper该映射器在第25行设置。 第27和28行分别设置了我们的自定义分区程序和组比较器以确保键和值到达化简器的顺序并使用正确的键正确地对值进行分组。 通过使用分区程序和分组比较器我们知道第一个值属于第一个键并且可以用于将Iterable包含的所有其他值连接到给定键的reduce()方法中。 现在是时候考虑一对多联接了。 一对多加入 好消息是到目前为止我们已经完成了所有工作实际上我们可以使用代码执行一对多连接。 对于一对多联接我们可以考虑两种方法1一个包含单个记录的小文件另一个包含具有相同键的多个记录的文件以及2同样具有单个记录的小文件但是N每个文件包含与第一个文件匹配的记录的文件数。 主要区别在于采用第一种方法时除了前两个键的联接之外值的顺序将是未知的。 但是使用第二种方法我们将“标记”每个联接文件以便我们可以控制所有联接值的顺序。 对于我们的示例第一个文件将保留为我们的GUID名称-地址文件并且我们将拥有3个其他文件其中将包含汽车雇主和工作描述记录。 这可能不是最现实的情况但将用于演示。 以下是在进行联接之前数据外观的示例 //The single person records
cdd8dde3-0349-4f0d-b97a-7ae84b687f9c,Esther,Garner,4071 Haven Lane,Okemos,MI
81a43486-07e1-4b92-b92b-03d0caa87b5f,Timothy,Duncan,753 Stadium Drive,Taunton,MA
aef52cf1-f565-4124-bf18-47acdac47a0e,Brett,Ramsey,4985 Shinn Street,New York,NY
//Automobile records
cdd8dde3-0349-4f0d-b97a-7ae84b687f9c,2003 Holden Cruze
81a43486-07e1-4b92-b92b-03d0caa87b5f,2012 Volkswagen T5
aef52cf1-f565-4124-bf18-47acdac47a0e,2009 Renault Trafic
//Employer records
cdd8dde3-0349-4f0d-b97a-7ae84b687f9c,Creative Wealth
81a43486-07e1-4b92-b92b-03d0caa87b5f,Susies Casuals
aef52cf1-f565-4124-bf18-47acdac47a0e,Super Saver Foods
//Job Description records
cdd8dde3-0349-4f0d-b97a-7ae84b687f9c,Data entry clerk
81a43486-07e1-4b92-b92b-03d0caa87b5f,Precision instrument and equipment repairer
aef52cf1-f565-4124-bf18-47acdac47a0e,Gas and water service dispatcher一对多加入结果 现在让我们看一下一对多联接结果的示例使用上面的相同值来辅助比较 cdd8dde3-0349-4f0d-b97a-7ae84b687f9c,Esther,Garner,4071 Haven Lane,Okemos,MI,2003 Holden Cruze,Creative Wealth,Data entry clerk
81a43486-07e1-4b92-b92b-03d0caa87b5f,Timothy,Duncan,753 Stadium Drive,Taunton,MA,2012 Volkswagen T5,Susies Casuals,Precision instrument and equipment repairer
aef52cf1-f565-4124-bf18-47acdac47a0e,Brett,Ramsey,4985 Shinn Street,New York,NY,2009 Renault Trafic,Super Saver Foods,Gas and water service dispatcher 结果表明我们已经能够成功地以指定顺序连接多个值。 结论 我们已经成功演示了如何在MapReduce中执行约简边连接。 即使该方法并不太复杂我们也可以看到在Hadoop中执行联接可能涉及编写大量代码。 虽然学习联接的工作方式是一项有用的练习但是在大多数情况下使用Hive或Pig这样的工具联接数据要好得多。 谢谢你的时间。 资源资源 Jimmy Lin和Chris Dyer 使用MapReduce进行的数据密集型处理 Hadoop Tom White 的权威指南 来自博客的源代码和测试 爱德华·卡普里奥洛Edward Capriolo迪恩·沃普勒Dean Wampler和杰森·卢瑟格伦Jason Rutherglen的编程蜂巢 通过Alan Gates对Pig进行编程 Hadoop API MRUnit用于单元测试Apache Hadoop映射减少工作 参考 MapReduce算法–了解数据 是我们的JCG合作伙伴 Bill Bejeck在“ 随机思考编码”博客上的第1部分 。 翻译自: https://www.javacodegeeks.com/2013/07/mapreduce-algorithms-understanding-data-joins-part-1.html
