首页 文章 精选 留言 我的

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

2019-01-27 607

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
上一篇 下一篇
优秀的个人博客,低调大师

微信关注我们

原文链接：https://yq.aliyun.com/articles/689111

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

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

mrjob是由Yelp创建的Python MapReduce库，它封装了Hadoop流，允许MapReduce应用程序以更加Pythonic的方式编写。 mrjob用纯Python编写多步MapReduce作业。使用mrjob编写的MapReduce作业可以在本地测试，在Hadoop集群上运行，或使用Amazon Elastic　MapReduce(EMR)在云中运行。 使用mrjob编写MapReduce应用程序有许多好处： mrjob目前是非常活跃的框架，每周都有多次提交。 mrjob拥有丰富的文档。 可以在不安装Hadoop的情况下执行和测试mrjob应用程序，在部署到Hadoop集群之前就可开发和测试。 mrjob允许MapReduce应用程序在单个类中编写，而不是为mapper和reducer编写单独的程序。 虽然mrjob是很好的解决方案，但它确实有它的缺点。 mrjob是简化的，因此它不会提供与其他API提供的Hadoop相同级别的访问权限。 mrjob不使用typedbytes，因此其他库可能更快。 安装 $ pip install mrjob 参考资料 python测...
2019-01-28
713
上一篇

[雪峰磁针石博客]大数据Hadoop工具python教程２-python访问HDFS

https://pypi.org/project/hdfs3　已经不维护PyArrowhttps://pypi.org/project/hdfs/https://pypi.org/project/snakebite/　python2中比较好，对python3支持不好。 hdfs和PyArrow比较常用，这里以hdfs为例： 　快速入门 from hdfs import InsecureClient client = InsecureClient('http://localhost:50070', user='hduser_') fs_folders_list = client.list("/") print(fs_folders_list) with client.read('/user/hduser/input.txt', enco
2019-01-28
666
下一篇

相关文章

发表评论

资源下载

更多资源
Mario

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

时间: 2025-12-12 大小: 211.28KB 下载: 54次
腾讯云软件源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

时间: 2026-03-01 大小: 1MB 下载: 6次
Spring

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

时间: 2025-12-17 大小: 5MB 下载: 1次
Rocky Linux

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

时间: 2026-03-02 大小: 1.42GB 下载: 3次
手机查看

扫码在手机上查看文章

扫描二维码，手机阅读更方便

热门标签
SpringBoot4 Redis Docker DeepSeek SpringCloud OpenAi Jdk25 Rocky Nacos3 Service Mesh Kubernetes HarmonyOS6 Gemini3

欢迎您来访！

登录后，您将获得更多功能！
热门文章
联系我们

有任何问题或合作意向欢迎联系我们

Email: 99873273@qq.com

QQ: 99873273

用户登录
用户注册