diff --git a/notes/Hadoop-MapReduce.md b/notes/Hadoop-MapReduce.md index aa76d36..10b0e94 100644 --- a/notes/Hadoop-MapReduce.md +++ b/notes/Hadoop-MapReduce.md @@ -4,13 +4,12 @@ 一、MapReduce 概述
二、MapReduce 编程模型简述
三、MapReduce 编程模型详述
-    3.1 InputFormat & RecordReaders
-    3.2 combiner
-    3.3 partitioner
-    3.4 sort & combiner
四、MapReduce 词频统计案例
+五、词频统计案例进阶
+ + ## 一、MapReduce 概述 Hadoop MapReduce是一个分布式计算框架，用于编写应用程序，以可靠，容错的方式在大型集群上并行处理大量数据（多为TB级别数据集）。 @@ -75,11 +74,23 @@ public abstract class InputFormat { ### 3.2 combiner -combiner是map运算后的可选操作，其实际上是一个本地化的reduce操作，它主要是在map计算出中间文件后做一个简单的合并重复key值的操作。例如我们对文件里的单词频率做统计，map计算时候如果碰到一个hadoop的单词就会记录为1，但是这篇文章里hadoop可能会出现n多次，那么map输出文件冗余就会很多，因此在reduce计算前对相同的key做一个合并操作，那么文件会变小。这样就提高了宽带的传输效率，因为hadoop计算的宽带资源往往是计算的瓶颈也是最为宝贵的资源。 +combiner是map运算后的可选操作，其实际上是一个本地化的reduce操作，它主要是在map计算出中间文件后做一个简单的合并重复key值的操作。 + +例如我们对文件里的单词频率做统计，map计算时候如果碰到一个hadoop的单词就会记录为1，但是这篇文章里hadoop可能会出现n多次，那么map输出文件冗余就会很多，因此在reduce计算前对相同的key做一个合并操作，那么文件会变小。这样就提高了宽带的传输效率，因为hadoop计算的宽带资源往往是计算的瓶颈也是最为宝贵的资源。 但并非所有场景都适合使用combiner，使用它的原则是combiner的输入不会影响到reduce计算的最终输入，例如：如果计算只是求总数，最大值，最小值可以使用combiner，但是做平均值计算使用combiner的话，最终的reduce计算结果就会出错。 -
+不使用combiner的情况： + +
+ +使用combiner的情况： + +
+ + + +可以看到使用combiner的时候，需要传输到reducer中的数据由12keys，降低到10keys。降低的幅度取决于你keys的重复率，后文词频统计案例可以直观演示用combiner降低数百倍的传输量。 ### 3.3 partitioner @@ -99,6 +110,364 @@ Merge是怎样的？如“aaa”从某个map task读取过来时值是5，从另 ## 四、MapReduce 词频统计案例 +### 4.1 项目简介 + +这里给出一个经典的案例：词频统计。统计如下样本数据中每个单词出现的次数。 + +```properties +Spark HBase +Hive Flink Storm Hadoop HBase Spark +Flink +HBase Storm +HBase Hadoop Hive Flink +HBase Flink Hive Storm +Hive Flink Hadoop +HBase Hive +Hadoop Spark HBase Storm +HBase Hadoop Hive Flink +HBase Flink Hive Storm +Hive Flink Hadoop +HBase Hive +``` + +为方便大家开发，我在项目源码中放置了一个工具类`WordCountDataUtils`，用于产生词频统计样本文件： + ++ 支持产生样本文件到本地，适用于本地测试； ++ 支持产生样本文件并直接输出到HDFS，适用于提交到服务器测试； + +> 本篇文章所有源码下载地址：[hadoop-word-count](https://github.com/heibaiying/BigData-Notes/tree/master/code/Hadoop/hadoop-word-count) + +```java +/** + * 产生词频统计模拟数据 + */ +public class WordCountDataUtils { + + public static final List WORD_LIST = Arrays.asList("Spark", "Hadoop", "HBase", "Storm", "Flink", "Hive"); + + + /** + * 模拟产生词频数据 + * + * @return 词频数据 + */ + private static String generateData() { + StringBuilder builder = new StringBuilder(); + for (int i = 0; i < 1000; i++) { + Collections.shuffle(WORD_LIST); + Random random = new Random(); + int endIndex = random.nextInt(WORD_LIST.size()) % (WORD_LIST.size()) + 1; + String line = StringUtils.join(WORD_LIST.toArray(), "\t", 0, endIndex); + builder.append(line).append("\n"); + } + return builder.toString(); + } + + + /** + * 模拟产生词频数据并输出到本地 + * + * @param outputPath 输出文件路径 + */ + private static void generateDataToLocal(String outputPath) { + try { + java.nio.file.Path path = Paths.get(outputPath); + if (Files.exists(path)) { + Files.delete(path); + } + Files.write(path, generateData().getBytes(), StandardOpenOption.CREATE); + } catch (IOException e) { + e.printStackTrace(); + } + } + + /** + * 模拟产生词频数据并输出到HDFS + * + * @param hdfsUrl HDFS地址 + * @param user hadoop用户名 + * @param outputPathString 存储到HDFS上的路径 + */ + private static void generateDataToHDFS(String hdfsUrl, String user, String outputPathString) { + FileSystem fileSystem = null; + try { + fileSystem = FileSystem.get(new URI(hdfsUrl), new Configuration(), user); + Path outputPath = new Path(outputPathString); + if (fileSystem.exists(outputPath)) { + fileSystem.delete(outputPath, true); + } + FSDataOutputStream out = fileSystem.create(outputPath); + out.write(generateData().getBytes()); + out.flush(); + out.close(); + fileSystem.close(); + } catch (Exception e) { + e.printStackTrace(); + } + } + + public static void main(String[] args) { + //generateDataToLocal("input.txt"); + generateDataToHDFS("hdfs://192.168.0.107:8020", "root", "/wordcount/input.txt"); + } +} +``` + +### 4.2 WordCountMapper + +**Mapper代码实现**： + +```java +/** + * 将每行数据按照指定分隔符进行拆分 + */ +public class WordCountMapper extends Mapper { + + @Override + protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { + String[] words = value.toString().split("\t"); + for (String word : words) { + context.write(new Text(word), new IntWritable(1)); + } + } + +} +``` + +**代码说明**： + +Splitting操作已由由Hadoop程序帮我们完成的，WordCountMapper对于下图的Mapping操作，这里WordCountMapper继承自Mapper类，这是一个泛型类，定义如下： + +```java +public class Mapper { + ...... +} +``` + ++ KEYIN : mapping输入的key的数据类型，即每行的偏移量（每行第一个字符在文本中的位置），Long类型，对应Hadoop中的LongWritable类型； ++ VALUEIN : mappin输入的value的数据类型，即每行数据；String类型，对应Hadoop中Text类型； ++ KEYOUT ：mapping输出的key的数据类型，即每个单词；String类型，对应Hadoop中Text类型； ++ VALUEOUT：mapping输出的value的数据类型，即每个单词出现的次数；这里用int类型，对应Hadoop中IntWritable类型； + +在MapReduce中必须使用Hadoop定义的类型，因为Hadoop预定义的类型都是可序列化，可比较的，所有类型均实现了`WritableComparable`接口。 + +
+ +### 4.3 WordCountReducer + +**Reducer代码实现**： + +```java +/** + * 进行词频统计 + */ +public class WordCountReducer extends Reducer { + + @Override + protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { + int count = 0; + for (IntWritable value : values) { + count += value.get(); + } + context.write(key, new IntWritable(count)); + } +} +``` + +**代码说明**： + +这里的key显然就是每个单词，这里的values是一个可迭代的数据类型，因为shuffling输出的数据实际上是下图中所示的这样的，即`key，(1,1,1,1,1,1,1,.....)`。values是可迭代的。 + +
+ +### 4.4 WordCountApp + +组装MapReduce作业，并提交到服务器运行，代码如下： + +```java + +/** + * 组装作业 并提交到集群运行 + */ +public class WordCountApp { + + + // 这里为了直观显示参数 使用了硬编码，实际开发中可以通过外部传参 + private static final String HDFS_URL = "hdfs://192.168.0.107:8020"; + private static final String HADOOP_USER_NAME = "root"; + + public static void main(String[] args) throws Exception { + + // 文件输入路径和输出路径由外部传参指定 + if (args.length < 2) { + System.out.println("Input and output paths are necessary!"); + return; + } + + // 需要指明hadoop用户名，否则在HDFS上创建目录时可能会抛出权限不足的异常 + System.setProperty("HADOOP_USER_NAME", HADOOP_USER_NAME); + + Configuration configuration = new Configuration(); + // 指明HDFS的地址 + configuration.set("fs.defaultFS", HDFS_URL); + + // 创建一个Job + Job job = Job.getInstance(configuration); + + // 设置运行的主类 + job.setJarByClass(WordCountApp.class); + + // 设置Mapper和Reducer + job.setMapperClass(WordCountMapper.class); + job.setReducerClass(WordCountReducer.class); + + // 设置Mapper输出key和value的类型 + job.setMapOutputKeyClass(Text.class); + job.setMapOutputValueClass(IntWritable.class); + + // 设置Reducer输出key和value的类型 + job.setOutputKeyClass(Text.class); + job.setOutputValueClass(IntWritable.class); + + // 如果输出目录已经存在，则必须先删除，否则重复运行程序时会抛出异常 + FileSystem fileSystem = FileSystem.get(new URI(HDFS_URL), configuration, HADOOP_USER_NAME); + Path outputPath = new Path(args[1]); + if (fileSystem.exists(outputPath)) { + fileSystem.delete(outputPath, true); + } + + // 设置作业输入文件和输出文件的路径 + FileInputFormat.setInputPaths(job, new Path(args[0])); + FileOutputFormat.setOutputPath(job, outputPath); + + // 将作业提交到群集并等待它完成，参数设置为true代表打印显示对应的进度 + boolean result = job.waitForCompletion(true); + + // 关闭之前创建的fileSystem + fileSystem.close(); + + // 根据作业结果,终止当前运行的Java虚拟机,退出程序 + System.exit(result ? 0 : -1); + + } +} +``` + +这里说明一下：`setMapOutputKeyClass`和`setOutputValueClass`控制reducer函数的输出类型。map函数的输出类型默认情况下和reducer函数式相同的，如果不同，则必须通过`setMapOutputKeyClass`和`setMapOutputValueClass`进行设置。 + +### 4.5 提交到服务器运行 + +在实际开发中，可以在本机配置hadoop开发环境，直接运行`main`方法既可。这里主要介绍一下打包提交到服务器运行： + +由于本项目没有使用除Hadoop外的第三方依赖，直接打包即可： + +```shell +# mvn clean package +``` + +使用以下命令运行作业： + +```shell +hadoop jar /usr/appjar/hadoop-word-count-1.0.jar \ +com.heibaiying.WordCountApp \ +/wordcount/input.txt /wordcount/output/WordCountApp +``` + +作业完成后查看HDFS上生成目录： + +```shell +# 查看目录 +hadoop fs -ls /wordcount/output/WordCountApp + +# 查看统计结果 +hadoop fs -cat /wordcount/output/WordCountApp/part-r-00000 +``` + +
+ + + +## 五、词频统计案例进阶 + +### 5.1 combiner + +#### 1. combiner的代码实现 + +combiner的代码实现比较简单，只要在组装作业时，添加下面一行代码即可 + +```java +// 设置Combiner +job.setCombinerClass(WordCountReducer.class); +``` + +#### 2. 测试结果 + +加入combiner后统计结果是不会有变化的，但是我们可以从打印的日志看出combiner的效果： + +没有加入combiner的打印日志： + +
+ +加入combiner后的打印日志如下。 + +
+ +这里我们只有一个输入文件并且小于128M，所以只有一个Map进行处理，可以看到经过combiner后，records由3519降低为6（样本中单词种类就只有6个），这一点从图中日志的`reduce input records`参数也可以看出来。在这个用例中combiner的效果就非常明显。 + +### 5.2 Partitioner + +#### 1. 默认Partitioner规则 + +这里假设有个需求：将不同单词的统计结果输出到不同文件。这种需求实际上比较常见，比如统计产品的销量时，需要将结果按照产品分类输出。 + +要实现这个功能，就需要用到自定义Partitioner，这里我们先说一下默认的分区规则：在构建job时候，如果不指定，默认的使用的是`HashPartitioner`，其实现如下： + +```java +public class HashPartitioner extends Partitioner { + + public int getPartition(K key, V value, + int numReduceTasks) { + return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks; + } + +} +``` + +对key进行哈希散列并对`numReduceTasks`取余，这里由于`numReduceTasks`默认值为1，所以我们之前的统计结果都输出到同一个文件中。 + +#### 2. 自定义Partitioner + +这里我们继承`Partitioner`自定义分区规则，这里按照单词进行分区： + +```java +/** + * 自定义partitioner,按照单词分区 + */ +public class CustomPartitioner extends Partitioner { + + public int getPartition(Text text, IntWritable intWritable, int numPartitions) { + return WordCountDataUtils.WORD_LIST.indexOf(text.toString()); + } +} +``` + +并在构建job时候指定使用我们自己的分区规则，并设置reduce的个数： + +```java +// 设置自定义分区规则 +job.setPartitionerClass(CustomPartitioner.class); +// 设置reduce个数 +job.setNumReduceTasks(WordCountDataUtils.WORD_LIST.size()); +``` + + + +#### 3. 测试结果 + +测试结果如下，分别生成6个文件，每个文件中为对应单词的统计结果。 + +
+ @@ -107,6 +476,7 @@ Merge是怎样的？如“aaa”从某个map task读取过来时值是5，从另 1. [分布式计算框架MapReduce](https://zhuanlan.zhihu.com/p/28682581) 2. [Apache Hadoop 2.9.2 > MapReduce Tutorial](http://hadoop.apache.org/docs/stable/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html) +3. [MapReduce - Combiners]( https://www.tutorialscampus.com/tutorials/map-reduce/combiners.htm) diff --git a/notes/installation/虚拟机静态IP及多IP配置.md b/notes/installation/虚拟机静态IP及多IP配置.md new file mode 100644 index 0000000..9c613dd --- /dev/null +++ b/notes/installation/虚拟机静态IP及多IP配置.md @@ -0,0 +1,116 @@ +# 虚拟机静态IP及多IP配置 + +> 虚拟机环境：centos 7.6 + + + + + + + +## 一、虚拟机静态IP配置 + +### 1. 查看当前网卡名称 + +​ 使用`ifconfig`，本机网卡名称为`enp0s3` + +
+ +### 2. 编辑网络配置文件 + +```shell +# vim /etc/sysconfig/network-scripts/ifcfg-enp0s3 +``` + +添加如下网络配置： + ++ IPADDR需要和宿主机同一个网段； ++ GATEWAY保持和宿主机一致； + +```properties +BOOTPROTO=static +IPADDR=192.168.0.107 +NETMASK=255.255.255.0 +GATEWAY=192.168.0.1 +DNS1=192.168.0.1 +ONBOOT=yes +``` + +我的主机配置： + +
+ +修改后完整配置如下： + +```properties +
TYPE=Ethernet +PROXY_METHOD=none +BROWSER_ONLY=no +BOOTPROTO=static +IPADDR=192.168.0.107 +NETMASK=255.255.255.0 +GATEWAY=192.168.0.1 +BROADCAST=192.168.0.255 +DNS1=192.168.0.1 +DEFROUTE=yes +IPV4_FAILURE_FATAL=no +IPV6INIT=yes +IPV6_AUTOCONF=yes +IPV6_DEFROUTE=yes +IPV6_FAILURE_FATAL=no +IPV6_ADDR_GEN_MODE=stable-privacy +NAME=enp0s3 +UUID=03d45df1-8514-4774-9b47-fddd6b9d9fca +DEVICE=enp0s3 +ONBOOT=yes +``` + +### 3. 重启网络服务 + +```shell +# systemctl restart network +``` + + + +## 二、虚拟机多个静态IP配置 + +这里说一下多个静态IP的使用场景：主要是针对同一台电脑在经常在不同网络环境使用（办公，家庭，学习等），配置好多个IP后，在`hosts`文件中映射到同一个主机名，这样在不同网络中就可以直接启动Hadoop等软件。 + +### 1. 配置多网卡 + +这里我是用的虚拟机是virtualBox，开启多网卡配置方式如下： + +
+ +### 2. 查看网卡名称 + +使用`ifconfig`，查看第二块网卡名称，这里我的名称为`enp0s8`。 + +
+ +### 3. 配置第二块网卡 + +开启多网卡后并不会自动生成配置文件，需要拷贝`ifcfg-enp0s3`进行修改。 + +```shell +# cp ifcfg-enp0s3 ifcfg-enp0s8 +``` + +静态IP配置方法如上，这里不再赘述。除了静态IP参数外，以下三个参数还需要修改，UUID必须与`ifcfg-enp0s3`中的不一样： + +```shell +NAME=enp0s8 +UUID=03d45df1-8514-4774-9b47-fddd6b9d9fcb +DEVICE=enp0s8 +``` + +### 4. 重启网络服务器 + +```shell +# systemctl restart network +``` + diff --git a/notes/installation/虚拟机静态IP配置.md b/notes/installation/虚拟机静态IP配置.md deleted file mode 100644 index a10db6a..0000000 --- a/notes/installation/虚拟机静态IP配置.md +++ /dev/null @@ -1,38 +0,0 @@ -# 虚拟机静态IP配置 - -> 虚拟机环境：centos 7.6 - - - -### 1. 查看当前网卡名称 - -​ 本机网卡名称为`enp0s3` - -
- -### 2. 编辑网络配置文件 - -```shell -# vim /etc/sysconfig/network-scripts/ifcfg-enp0s3 -``` - -添加如下网络配置，指明静态IP和DNS： - -```shell -BOOTPROTO=static -IPADDR=192.168.200.226 -NETMASK=255.255.255.0 -GATEWAY=192.168.200.254 -DNS1=114.114.114.114 -``` - -修改后完整配置如下： - -
- -### 3. 重启网络服务 - -```shell -# systemctl restart network -``` - diff --git a/pictures/hadoop-code-mapping.png b/pictures/hadoop-code-mapping.png new file mode 100644 index 0000000..c903651 Binary files /dev/null and b/pictures/hadoop-code-mapping.png differ diff --git a/pictures/hadoop-code-partitation.png b/pictures/hadoop-code-partitation.png new file mode 100644 index 0000000..886e127 Binary files /dev/null and b/pictures/hadoop-code-partitation.png differ diff --git a/pictures/hadoop-code-reducer.png b/pictures/hadoop-code-reducer.png new file mode 100644 index 0000000..cc59314 Binary files /dev/null and b/pictures/hadoop-code-reducer.png differ diff --git a/pictures/hadoop-no-combiner.png b/pictures/hadoop-no-combiner.png index b946a8b..3fd9fd8 100644 Binary files a/pictures/hadoop-no-combiner.png and b/pictures/hadoop-no-combiner.png differ diff --git a/pictures/ipconfig.png b/pictures/ipconfig.png new file mode 100644 index 0000000..cfdf636 Binary files /dev/null and b/pictures/ipconfig.png differ diff --git a/pictures/mapreduce-with-combiners.png b/pictures/mapreduce-with-combiners.png new file mode 100644 index 0000000..498e012 Binary files /dev/null and b/pictures/mapreduce-with-combiners.png differ diff --git a/pictures/mapreduce-without-combiners.png b/pictures/mapreduce-without-combiners.png new file mode 100644 index 0000000..8394883 Binary files /dev/null and b/pictures/mapreduce-without-combiners.png differ diff --git a/pictures/mutli-net-ip.png b/pictures/mutli-net-ip.png new file mode 100644 index 0000000..a7b0d44 Binary files /dev/null and b/pictures/mutli-net-ip.png differ diff --git a/pictures/virtualbox-multi-network.png b/pictures/virtualbox-multi-network.png new file mode 100644 index 0000000..1aec7ce Binary files /dev/null and b/pictures/virtualbox-multi-network.png differ