[雪峰磁针石博客]大数据Hadoop工具python教程3-MapReduce-阿里云开发者社区

开发者社区> python人工智能命理> 正文

[雪峰磁针石博客]大数据Hadoop工具python教程3-MapReduce

简介: MapReduce是一种编程模型,通过将工作分成独立的任务并在一组机器上并行执行任务,可以处理和生成大量数据。 MapReduce编程风格的灵感来自函数式编程结构map和reduce,它们通常用于处理数据列表。
+关注继续查看

MapReduce是一种编程模型,通过将工作分成独立的任务并在一组机器上并行执行任务,可以处理和生成大量数据。 MapReduce编程风格的灵感来自函数式编程结构map和reduce,它们通常用于处理数据列表。在高层MapReduce程序将输入数据元素列表转换为输出数据元素列表两次,一次在映射阶段,一次在还原阶段。

本章首先介绍MapReduce编程模型,并描述数据如何流经模型的不同阶段。然后示例如何使用Python编写MapReduce作业。

数据流

MapReduce框架由三个主要阶段组成:map,shuffle和sort,以及reduce。

图片.png

  • map

在映射阶段,mapper函数处理一系列键值对。映射器按顺序处理键值对,产生零个或多个输出键值对。

比如将句子转换为单词。输入是包含句子的字符串,映射器将句子拆分为单词并输出单词。

图片.png

  • Shuffle和Sort

映射阶段的中间输出将移动到reducer。将输出从映射器移动到reducer的过程称为洗(shuffling)。

Shuffling由分区函数处理,称为partitioner。partitioner用于控制从映射器到reducer的键值对的流动。reducer知道映射器的输出键和reducer的数量,返回预期的reducer的索引。partitioner程序确保将同一键的所有值发送到同一reducer。默认分区程序是基于哈希的。它计算映射器输出键的哈希值,并根据此结果分配分区。

reducers开始处理数据之前的最后阶段是排序过程。在呈现给reducer之前,每个分区的中间键和值都由Hadoop框架排序。

  • Reduce

图片.png

在reducer阶段,值的迭代器被提供给称为reducer的函数。迭代器把值提供给reducer,这些值是唯一键的一组非唯一值。 reducer聚合每个唯一键的值,并产生零个或多个输出键值对。
数据流|

比如对键的所有值求和。此reducer的输入是键的所有值,reducer对所有值求和。然后,reducer输出键值对中包含输入键和输入键值的总和。

图片.png

参考资料

Hadoop流

Hadoop流是与Hadoop发行版一起打包的工具,它允许使用任何可执行文件创建MapReduce作业作为映射器和reducer。 Hadoop流实用程序支持Python,shell。

mapper和reducer都是可执行文件,它们从标准输入(stdin),逐行读取输入,并将写入标准输出(stdout)。 Hadoop流公国创建MapReduce作业,将作业提交到集群,并监视其进度直到完成。

mapper初始化时,每个映射任务都会将指定的可执行文件作为单独的进程启动。映射器读取输入文件,并通过stdin将每行显示给可执行文件。在可执行文件处理每行输入后,映射器从stdout收集输出并将每一行转换为键值对。键由第一个制表符前面的行部分组成,值由第一个制表符后面的行部分组成。如果一行不包含制表符,则整行被视为键,值为null。

初始化reducer时,每个reduce任务都会将指定的可执行文件作为单独的进程启动。 reducer将输入键值对转换为通过stdin呈现给可执行文件的行。

reducer从stdout收集可执行文件的结果,并将每一行转换为键值对。与映射器类似,可执行文件通过制表符分隔键和值来指定键值对。

下面我们用python来模拟Hadoop流工具。

mapper.py:在WordCount的map阶段实现逻辑的Python程序。它从stdin读取数据,将行拆分为单词,并将每个单词的中间计数输出到stdout。

#!/usr/bin/env python
# https://github.com/china-testing/python-api-tesing
import sys

# Read each line from STDIN
for line in sys.stdin:

   # Get the words in each line
   words = line.split()

   # Generate the count for each word
   for word in words:

      # Write the key-value pair to STDOUT to be processed by the reducer.
      # The key is anything before the first tab character and the value is
      # anything after the first tab character.
      print('{0}\t{1}'.format(word, 1))

reducer.py是在WordCount的reduce阶段实现逻辑的Python程序。它从stdin中读取mapper.py的结果,对每个单词的出现次数求和,并将结果写入stdout。

#!/usr/bin/env python
import sys

curr_word = None
curr_count = 0

# Process each key-value pair from the mapper
for line in sys.stdin:

   # Get the key and value from the current line
   word, count = line.split('\t')

   # Convert the count to an int
   count = int(count)

   # If the current word is the same as the previous word, increment its
   # count, otherwise print the words count to STDOUT
   if word == curr_word:
      curr_count += count
   else: 

      # Write word and its number of occurrences as a key-value pair to STDOUT
      if curr_word:
         print('{0}\t{1}'.format(curr_word, curr_count))

      curr_word = word
      curr_count = count

# Output the count for the last word
if curr_word == word:
   print('{0}\t{1}'.format(curr_word, curr_count))

执行

$ echo 'jack be nimble jack be quick' | ./mapper.py | sort -t 1 | ./reducer.py
be    2
jack    2
nimble    1
quick    1

现在我们把'jack be nimble jack be quick'存成/home/hduser_/input2.txt,用hadoop来实现这一过程。

$ hdfs dfs -put /home/hduser_/input2.txt /user/hduser
$ $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.9.2.jar -files mapper.py,reducer.py -mapper mapper.py -reducer reducer.py -input /user/hduser/input2.txt -output /user/hduser/output
19/01/22 10:44:38 INFO Configuration.deprecation: session.id is deprecated. Instead, use dfs.metrics.session-id
19/01/22 10:44:38 INFO jvm.JvmMetrics: Initializing JVM Metrics with processName=JobTracker, sessionId=
19/01/22 10:44:38 INFO jvm.JvmMetrics: Cannot initialize JVM Metrics with processName=JobTracker, sessionId= - already initialized
19/01/22 10:44:38 ERROR streaming.StreamJob: Error Launching job : Output directory hdfs://localhost:54310/user/hduser/output already exists
Streaming Command Failed!
hduser_@andrew-PC:/home/andrew/code/HadoopWithPython/python/MapReduce/HadoopStreaming$ $HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/share/hadoop/tools/lib/hadoop-streaming-2.9.2.jar -files mapper.py,reducer.py -mapper mapper.py -reducer reducer.py -input /user/hduser/input2.txt -output /user/hduser/output2
19/01/22 10:44:45 INFO Configuration.deprecation: session.id is deprecated. Instead, use dfs.metrics.session-id
19/01/22 10:44:45 INFO jvm.JvmMetrics: Initializing JVM Metrics with processName=JobTracker, sessionId=
19/01/22 10:44:45 INFO jvm.JvmMetrics: Cannot initialize JVM Metrics with processName=JobTracker, sessionId= - already initialized
19/01/22 10:44:46 INFO mapred.FileInputFormat: Total input files to process : 1
19/01/22 10:44:46 INFO mapreduce.JobSubmitter: number of splits:1
19/01/22 10:44:46 INFO mapreduce.JobSubmitter: Submitting tokens for job: job_local208759810_0001
19/01/22 10:44:46 INFO mapred.LocalDistributedCacheManager: Localized file:/home/andrew/code/HadoopWithPython/python/MapReduce/HadoopStreaming/mapper.py as file:/app/hadoop/tmp/mapred/local/1548125086275/mapper.py
19/01/22 10:44:46 INFO mapred.LocalDistributedCacheManager: Localized file:/home/andrew/code/HadoopWithPython/python/MapReduce/HadoopStreaming/reducer.py as file:/app/hadoop/tmp/mapred/local/1548125086276/reducer.py
19/01/22 10:44:46 INFO mapreduce.Job: The url to track the job: http://localhost:8080/
19/01/22 10:44:46 INFO mapred.LocalJobRunner: OutputCommitter set in config null
19/01/22 10:44:46 INFO mapreduce.Job: Running job: job_local208759810_0001
19/01/22 10:44:46 INFO mapred.LocalJobRunner: OutputCommitter is org.apache.hadoop.mapred.FileOutputCommitter
19/01/22 10:44:46 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 1
19/01/22 10:44:46 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false
19/01/22 10:44:46 INFO mapred.LocalJobRunner: Waiting for map tasks
19/01/22 10:44:46 INFO mapred.LocalJobRunner: Starting task: attempt_local208759810_0001_m_000000_0
19/01/22 10:44:46 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 1
19/01/22 10:44:46 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false
19/01/22 10:44:46 INFO mapred.Task:  Using ResourceCalculatorProcessTree : [ ]
19/01/22 10:44:46 INFO mapred.MapTask: Processing split: hdfs://localhost:54310/user/hduser/input2.txt:0+29
19/01/22 10:44:46 INFO mapred.MapTask: numReduceTasks: 1
19/01/22 10:44:46 INFO mapred.MapTask: (EQUATOR) 0 kvi 26214396(104857584)
19/01/22 10:44:46 INFO mapred.MapTask: mapreduce.task.io.sort.mb: 100
19/01/22 10:44:46 INFO mapred.MapTask: soft limit at 83886080
19/01/22 10:44:46 INFO mapred.MapTask: bufstart = 0; bufvoid = 104857600
19/01/22 10:44:46 INFO mapred.MapTask: kvstart = 26214396; length = 6553600
19/01/22 10:44:46 INFO mapred.MapTask: Map output collector class = org.apache.hadoop.mapred.MapTask$MapOutputBuffer
19/01/22 10:44:46 INFO streaming.PipeMapRed: PipeMapRed exec [/home/andrew/code/HadoopWithPython/python/MapReduce/HadoopStreaming/./mapper.py]
19/01/22 10:44:46 INFO Configuration.deprecation: mapred.work.output.dir is deprecated. Instead, use mapreduce.task.output.dir
19/01/22 10:44:46 INFO Configuration.deprecation: map.input.start is deprecated. Instead, use mapreduce.map.input.start
19/01/22 10:44:46 INFO Configuration.deprecation: mapred.task.is.map is deprecated. Instead, use mapreduce.task.ismap
19/01/22 10:44:46 INFO Configuration.deprecation: mapred.task.id is deprecated. Instead, use mapreduce.task.attempt.id
19/01/22 10:44:46 INFO Configuration.deprecation: mapred.tip.id is deprecated. Instead, use mapreduce.task.id
19/01/22 10:44:46 INFO Configuration.deprecation: mapred.local.dir is deprecated. Instead, use mapreduce.cluster.local.dir
19/01/22 10:44:46 INFO Configuration.deprecation: map.input.file is deprecated. Instead, use mapreduce.map.input.file
19/01/22 10:44:46 INFO Configuration.deprecation: mapred.skip.on is deprecated. Instead, use mapreduce.job.skiprecords
19/01/22 10:44:46 INFO Configuration.deprecation: map.input.length is deprecated. Instead, use mapreduce.map.input.length
19/01/22 10:44:46 INFO Configuration.deprecation: mapred.job.id is deprecated. Instead, use mapreduce.job.id
19/01/22 10:44:46 INFO Configuration.deprecation: user.name is deprecated. Instead, use mapreduce.job.user.name
19/01/22 10:44:46 INFO Configuration.deprecation: mapred.task.partition is deprecated. Instead, use mapreduce.task.partition
19/01/22 10:44:46 INFO streaming.PipeMapRed: R/W/S=1/0/0 in:NA [rec/s] out:NA [rec/s]
19/01/22 10:44:46 INFO streaming.PipeMapRed: Records R/W=1/1
19/01/22 10:44:46 INFO streaming.PipeMapRed: MRErrorThread done
19/01/22 10:44:46 INFO streaming.PipeMapRed: mapRedFinished
19/01/22 10:44:46 INFO mapred.LocalJobRunner: 
19/01/22 10:44:46 INFO mapred.MapTask: Starting flush of map output
19/01/22 10:44:46 INFO mapred.MapTask: Spilling map output
19/01/22 10:44:46 INFO mapred.MapTask: bufstart = 0; bufend = 41; bufvoid = 104857600
19/01/22 10:44:46 INFO mapred.MapTask: kvstart = 26214396(104857584); kvend = 26214376(104857504); length = 21/6553600
19/01/22 10:44:46 INFO mapred.MapTask: Finished spill 0
19/01/22 10:44:46 INFO mapred.Task: Task:attempt_local208759810_0001_m_000000_0 is done. And is in the process of committing
19/01/22 10:44:46 INFO mapred.LocalJobRunner: Records R/W=1/1
19/01/22 10:44:46 INFO mapred.Task: Task 'attempt_local208759810_0001_m_000000_0' done.
19/01/22 10:44:46 INFO mapred.LocalJobRunner: Finishing task: attempt_local208759810_0001_m_000000_0
19/01/22 10:44:46 INFO mapred.LocalJobRunner: map task executor complete.
19/01/22 10:44:46 INFO mapred.LocalJobRunner: Waiting for reduce tasks
19/01/22 10:44:46 INFO mapred.LocalJobRunner: Starting task: attempt_local208759810_0001_r_000000_0
19/01/22 10:44:46 INFO output.FileOutputCommitter: File Output Committer Algorithm version is 1
19/01/22 10:44:46 INFO output.FileOutputCommitter: FileOutputCommitter skip cleanup _temporary folders under output directory:false, ignore cleanup failures: false
19/01/22 10:44:46 INFO mapred.Task:  Using ResourceCalculatorProcessTree : [ ]
19/01/22 10:44:46 INFO mapred.ReduceTask: Using ShuffleConsumerPlugin: org.apache.hadoop.mapreduce.task.reduce.Shuffle@78674b51
19/01/22 10:44:46 INFO reduce.MergeManagerImpl: MergerManager: memoryLimit=334338464, maxSingleShuffleLimit=83584616, mergeThreshold=220663392, ioSortFactor=10, memToMemMergeOutputsThreshold=10
19/01/22 10:44:46 INFO reduce.EventFetcher: attempt_local208759810_0001_r_000000_0 Thread started: EventFetcher for fetching Map Completion Events
19/01/22 10:44:46 INFO reduce.LocalFetcher: localfetcher#1 about to shuffle output of map attempt_local208759810_0001_m_000000_0 decomp: 55 len: 59 to MEMORY
19/01/22 10:44:46 INFO reduce.InMemoryMapOutput: Read 55 bytes from map-output for attempt_local208759810_0001_m_000000_0
19/01/22 10:44:46 INFO reduce.MergeManagerImpl: closeInMemoryFile -> map-output of size: 55, inMemoryMapOutputs.size() -> 1, commitMemory -> 0, usedMemory ->55
19/01/22 10:44:46 INFO reduce.EventFetcher: EventFetcher is interrupted.. Returning
19/01/22 10:44:46 INFO mapred.LocalJobRunner: 1 / 1 copied.
19/01/22 10:44:46 INFO reduce.MergeManagerImpl: finalMerge called with 1 in-memory map-outputs and 0 on-disk map-outputs
19/01/22 10:44:46 INFO mapred.Merger: Merging 1 sorted segments
19/01/22 10:44:46 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 50 bytes
19/01/22 10:44:46 INFO reduce.MergeManagerImpl: Merged 1 segments, 55 bytes to disk to satisfy reduce memory limit
19/01/22 10:44:46 INFO reduce.MergeManagerImpl: Merging 1 files, 59 bytes from disk
19/01/22 10:44:46 INFO reduce.MergeManagerImpl: Merging 0 segments, 0 bytes from memory into reduce
19/01/22 10:44:46 INFO mapred.Merger: Merging 1 sorted segments
19/01/22 10:44:46 INFO mapred.Merger: Down to the last merge-pass, with 1 segments left of total size: 50 bytes
19/01/22 10:44:46 INFO mapred.LocalJobRunner: 1 / 1 copied.
19/01/22 10:44:46 INFO streaming.PipeMapRed: PipeMapRed exec [/home/andrew/code/HadoopWithPython/python/MapReduce/HadoopStreaming/./reducer.py]
19/01/22 10:44:46 INFO Configuration.deprecation: mapred.job.tracker is deprecated. Instead, use mapreduce.jobtracker.address
19/01/22 10:44:46 INFO Configuration.deprecation: mapred.map.tasks is deprecated. Instead, use mapreduce.job.maps
19/01/22 10:44:46 INFO streaming.PipeMapRed: R/W/S=1/0/0 in:NA [rec/s] out:NA [rec/s]
19/01/22 10:44:46 INFO streaming.PipeMapRed: Records R/W=6/1
19/01/22 10:44:46 INFO streaming.PipeMapRed: MRErrorThread done
19/01/22 10:44:46 INFO streaming.PipeMapRed: mapRedFinished
19/01/22 10:44:46 INFO mapred.Task: Task:attempt_local208759810_0001_r_000000_0 is done. And is in the process of committing
19/01/22 10:44:46 INFO mapred.LocalJobRunner: 1 / 1 copied.
19/01/22 10:44:46 INFO mapred.Task: Task attempt_local208759810_0001_r_000000_0 is allowed to commit now
19/01/22 10:44:46 INFO output.FileOutputCommitter: Saved output of task 'attempt_local208759810_0001_r_000000_0' to hdfs://localhost:54310/user/hduser/output2/_temporary/0/task_local208759810_0001_r_000000
19/01/22 10:44:46 INFO mapred.LocalJobRunner: Records R/W=6/1 > reduce
19/01/22 10:44:46 INFO mapred.Task: Task 'attempt_local208759810_0001_r_000000_0' done.
19/01/22 10:44:46 INFO mapred.LocalJobRunner: Finishing task: attempt_local208759810_0001_r_000000_0
19/01/22 10:44:46 INFO mapred.LocalJobRunner: reduce task executor complete.
19/01/22 10:44:47 INFO mapreduce.Job: Job job_local208759810_0001 running in uber mode : false
19/01/22 10:44:47 INFO mapreduce.Job:  map 100% reduce 100%
19/01/22 10:44:47 INFO mapreduce.Job: Job job_local208759810_0001 completed successfully
19/01/22 10:44:47 INFO mapreduce.Job: Counters: 35
    File System Counters
        FILE: Number of bytes read=273356
        FILE: Number of bytes written=1217709
        FILE: Number of read operations=0
        FILE: Number of large read operations=0
        FILE: Number of write operations=0
        HDFS: Number of bytes read=58
        HDFS: Number of bytes written=29
        HDFS: Number of read operations=13
        HDFS: Number of large read operations=0
        HDFS: Number of write operations=4
    Map-Reduce Framework
        Map input records=1
        Map output records=6
        Map output bytes=41
        Map output materialized bytes=59
        Input split bytes=97
        Combine input records=0
        Combine output records=0
        Reduce input groups=4
        Reduce shuffle bytes=59
        Reduce input records=6
        Reduce output records=4
        Spilled Records=12
        Shuffled Maps =1
        Failed Shuffles=0
        Merged Map outputs=1
        GC time elapsed (ms)=0
        Total committed heap usage (bytes)=552599552
    Shuffle Errors
        BAD_ID=0
        CONNECTION=0
        IO_ERROR=0
        WRONG_LENGTH=0
        WRONG_MAP=0
        WRONG_REDUCE=0
    File Input Format Counters 
        Bytes Read=29
    File Output Format Counters 
        Bytes Written=29
19/01/22 10:44:47 INFO streaming.StreamJob: Output directory: /user/hduser/output2
$ hdfs dfs -cat  /user/hduser/output2/part-00000
be    2
jack    2
nimble    1
quick    1

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
史上最快! 10小时大数据入门实战(五)-分布式计算框架MapReduce
目录 1 MapReduce概述 2 MapReduce编程模型之通过wordcount词频统计分析案例入门 MapReduce执行流程 InputFormat OutputFormat OutputFormt接口决定了在哪里以及怎样持久化作业结果。
1032 0
Java基础-18总结Map,HashMap,HashMap与Hashtable区别,Collections工具类
你需要的是什么,直接评论留言。 获取更多资源加微信公众号“Java帮帮” (是公众号,不是微信好友哦) 还有“Java帮帮”今日头条号,技术文章与新闻,每日更新,欢迎阅读 学习交流请加Java帮帮交流QQ群553841695 分享是一种美德,分享更快乐! 1:Map(掌握) (1)将键映射到值的对象。一个映射不能包含重复的键;每个键最多只
2416 0
Python爬虫入门教程 31-100 36氪(36kr)数据抓取 scrapy
1. 36氪(36kr)数据----写在前面 今天抓取一个新闻媒体,36kr的文章内容,也是为后面的数据分析做相应的准备的,预计在12月底,爬虫大概写到50篇案例的时刻,将会迎来一个新的内容,系统的数据分析博文,记得关注哦~ 36kr 让一部分人先看到未来,而你今天要做的事情确实要抓取它的过去。
9249 0
Python+大数据计算平台,PyODPS架构手把手教你搭建
在2016年10月的云栖社区在线培训上,来自阿里云大数据事业部的秦续业分享了《双剑合壁——Python和大数据计算平台的结合实战》。他主要介绍了数据分析和机器学习的方法、DataFrame整体架构以及基础API、前端、后端、机器学习的具体实现方法。
16355 0
Progress进度查看器是一个可以显示Linux命令的工具
Progress进度查看器是一个可以显示Linux命令的工具
316 0
Directx11教程(42) 纹理映射(12)-简单的bump mapping
有时候,我们只有一个粗糙的模型,但是我们想渲染纹理细节,比如一个砖墙,我们如何在只有一个平面的时候,渲染出砖墙凹凸的效果。    比如只有这样的墙:   但是我们想要这样的效果: 怎么办呢?这时候,我们可以考虑对第一张图进行处理,生成它的法向图,存储在一张纹理中,生成...
674 0
+关注
python人工智能命理
提供书籍寻找服务,十元一本起,不成功不收费。 曾任职于腾讯、华为等公司,独立带队完成大型项目的全系统测试,担任过技术总监、测试经理以及测试专家职位。现任某主流公司自动化测试架构师。
107
文章
1
问答
文章排行榜
最热
最新
相关电子书
更多
文娱运维技术
立即下载
《SaaS模式云原生数据仓库应用场景实践》
立即下载
《看见新力量:二》电子书
立即下载