程序员经验分享-hwhale

mapreduce编程实例python-使用Python实现Hadoop MapReduce程序

2023-05-16

在这个实例中，我将会向大家介绍如何使用Python 为Hadoop编写一个简单的MapReduce

程序。

尽管Hadoop 框架是使用Java编写的但是我们仍然需要使用像C++、Python等语言来实现 Hadoop程序。尽管Hadoop官方网站给的示例程序是使用Jython编写并打包成Jar文件，这样显然造成了不便，其实，不一定非要这样来实现，我们可以使用Python与Hadoop 关联进行编程，看看位于/src/examples/python/WordCount.py 的例子，你将了解到我在说什么。

我们想要做什么？我们将编写一个简单的 MapReduce 程序，使用的是C-Python，而不是Jython编写后打包成jar包的程序。

我们的这个例子将模仿 WordCount 并使用Python来实现，例子通过读取文本文件来统计出单词的出现次数。结果也以文本形式输出，每一行包含一个单词和单词出现的次数，两者中间使用制表符来想间隔。

先决条件

编写这个程序之前，你学要架设好Hadoop 集群，这样才能不会在后期工作抓瞎。如果你没有架设好，那么在后面有个简明教程来教你在Ubuntu Linux 上搭建（同样适用于其他发行版linux、unix）

如何使用Hadoop Distributed File System (HDFS)在Ubuntu Linux 建立单节点的 Hadoop 集群

如何使用Hadoop Distributed File System (HDFS)在Ubuntu Linux 建立多节点的 Hadoop 集群

Python的MapReduce代码

使用Python编写MapReduce代码的技巧就在于我们使用了 HadoopStreaming 来帮助我们在Map 和 Reduce间传递数据通过STDIN (标准输入)和STDOUT (标准输出).我们仅仅使用Python的sys.stdin来输入数据，使用sys.stdout输出数据，这样做是因为HadoopStreaming会帮我们办好其他事。这是真的，别不相信！

Map: mapper.py

将下列的代码保存在/home/hadoop/mapper.py中，他将从STDIN读取数据并将单词成行分隔开，生成一个列表映射单词与发生次数的关系：

注意：要确保这个脚本有足够权限（chmod +x /home/hadoop/mapper.py）。#!/usr/bin/env python

import sys

# input comes from STDIN (standard input)

for line in sys.stdin:

# remove leading and trailing whitespace

line = line.strip()

# split the line into words

words = line.split()

# increase counters

for word in words:

# write the results to STDOUT (standard output);

# what we output here will be the input for the

# Reduce step, i.e. the input for reducer.py

#

# tab-delimited; the trivial word count is 1

print '%s\t%s' % (word, 1)

在这个脚本中，并不计算出单词出现的总数，它将输出 " 1" 迅速地，尽管可能会在输入中出现多次，计算是留给后来的Reduce步骤（或叫做程序）来实现。当然你可以改变下编码风格，完全尊重你的习惯。

Reduce: reducer.py

将代码存储在/home/hadoop/reducer.py 中，这个脚本的作用是从mapper.py 的STDIN中读取结果，然后计算每个单词出现次数的总和，并输出结果到STDOUT。

同样，要注意脚本权限：chmod +x /home/hadoop/reducer.py#!/usr/bin/env python

from operator import itemgetter

import sys

# maps words to their counts

word2count = {}

# input comes from STDIN

for line in sys.stdin:

# remove leading and trailing whitespace

line = line.strip()

# parse the input we got from mapper.py

word, count = line.split('\t', 1)

# convert count (currently a string) to int

try:

count = int(count)

word2count[word] = word2count.get(word, 0) + count

except ValueError:

# count was not a number, so silently

# ignore/discard this line

pass

# sort the words lexigraphically;

#

# this step is NOT required, we just do it so that our

# final output will look more like the official Hadoop

# word count examples

sorted_word2count = sorted(word2count.items(), key=itemgetter(0))

# write the results to STDOUT (standard output)

for word, count in sorted_word2count:

print '%s\t%s'% (word, count)

测试你的代码（cat data | map | sort | reduce）我建议你在运行MapReduce job测试前尝试手工测试你的mapper.py 和 reducer.py脚本，以免得不到任何返回结果

这里有一些建议，关于如何测试你的Map和Reduce的功能：——————————————————————————————————————————————

# very basic test

hadoop@ubuntu:~$ echo "foo foo quux labs foo bar quux" | /home/hadoop/mapper.py

foo 1

foo 1

quux 1

labs 1

foo 1

bar 1

——————————————————————————————————————————————

hadoop@ubuntu:~$ echo "foo foo quux labs foo bar quux" | /home/hadoop/mapper.py | sort | /home/hadoop/reducer.py

bar 1

foo 3

labs 1

——————————————————————————————————————————————

# using on[object Object]e of the ebooks as example input

# (see below on where to get the ebooks)

hadoop@ubuntu:~$ cat /tmp/gutenberg/20417-8.txt | /home/hadoop/mapper.py

The 1

Project 1

Gutenberg 1

EBook 1

of 1

[...]

(you get the idea)

quux 2

quux 1

——————————————————————————————————————————————

为了这个例子，我们将需要三种电子书：

下载他们，并使用us-ascii编码存储 解压后的文件，保存在临时目录，比如/tmp/gutenberg.hadoop@ubuntu:~$ ls -l /tmp/gutenberg/

total 3592

-rw-r--r-- 1 hadoop hadoop 674425 2007-01-22 12:56 20417-8.txt

-rw-r--r-- 1 hadoop hadoop 1423808 2006-08-03 16:36 7ldvc10.txt

-rw-r--r-- 1 hadoop hadoop 1561677 2004-11-26 09:48 ulyss12.txt

hadoop@ubuntu:~$在我们运行MapReduce job 前，我们需要将本地的文件复制到HDFS中：

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -copyFromLocal /tmp/gutenberg gutenberg

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -ls

Found 1 items

/user/hadoop/gutenberg

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -ls gutenberg

Found 3 items

/user/hadoop/gutenberg/20417-8.txt 674425

/user/hadoop/gutenberg/7ldvc10.txt 1423808

/user/hadoop/gutenberg/ulyss12.txt 1561677

现在，一切准备就绪，我们将在运行Python MapReduce job 在Hadoop集群上。像我上面所说的，我们使用的是

帮助我们传递数据在Map和Reduce间并通过STDIN和STDOUT，进行标准化输入输出。

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop jar contrib/streaming/hadoop-0.19.1-streaming.jar

-mapper /home/hadoop/mapper.py -reducer /home/hadoop/reducer.py -input gutenberg/*

-output gutenberg-output

在运行中，如果你想更改Hadoop的一些设置，如增加Reduce任务的数量，你可以使用“-hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop jar contrib/streaming/hadoop-0.19.1-streaming.jar

-mapper ...

一个重要的备忘是关于

这个任务将会读取HDFS目录下的HDFS目录下的

目录。

之前执行的结果如下：

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop jar contrib/streaming/hadoop-0.19.1-streaming.jar

-mapper /home/hadoop/mapper.py -reducer /home/hadoop/reducer.py -input gutenberg/*

-output gutenberg-output

additionalConfSpec_:null

null=@@@userJobConfProps_.get(stream.shipped.hadoopstreaming

packageJobJar: [/usr/local/hadoop-datastore/hadoop-hadoop/hadoop-unjar54543/]

[] /tmp/streamjob54544.jar tmpDir=null

[...] INFO mapred.FileInputFormat: Total input paths to process : 7

[...] INFO streaming.StreamJob: getLocalDirs(): [/usr/local/hadoop-datastore/hadoop-hadoop/mapred/local]

[...] INFO streaming.StreamJob: Running job: job_200803031615_0021

[...]

[...] INFO streaming.StreamJob: map 0% reduce 0%

[...] INFO streaming.StreamJob: map 43% reduce 0%

[...] INFO streaming.StreamJob: map 86% reduce 0%

[...] INFO streaming.StreamJob: map 100% reduce 0%

[...] INFO streaming.StreamJob: map 100% reduce 33%

[...] INFO streaming.StreamJob: map 100% reduce 70%

[...] INFO streaming.StreamJob: map 100% reduce 77%

[...] INFO streaming.StreamJob: map 100% reduce 100%

[...] INFO streaming.StreamJob: Job complete: job_200803031615_0021

[...] INFO streaming.StreamJob: Output: gutenberg-output hadoop@ubuntu:/usr/local/hadoop$

正如你所见到的上面的输出结果，Hadoop 同时还提供了一个基本的WEB接口显示统计结果和信息。

当Hadoop集群在执行时，你可以使用浏览器访问 ，如图：

检查结果是否输出并存储在HDFS目录下的中：

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -ls gutenberg-output

Found 1 items

/user/hadoop/gutenberg-output/part-00000 903193 2007-09-21 13:00

hadoop@ubuntu:/usr/local/hadoop$

可以使用 命令检查文件目录

hadoop@ubuntu:/usr/local/hadoop$ bin/hadoop dfs -cat gutenberg-output/part-00000

"(Lo)cra" 1

"1490 1

"1498," 1

"35" 1

"40," 1

"A 2

"AS-IS". 2

"A_ 1

"Absoluti 1

[...]

hadoop@ubuntu:/usr/local/hadoop$

注意比输出，上面结果的(")符号不是Hadoop插入的。

请参考：

http://www.michael-noll.com/wiki/Writing_An_Hadoop_MapReduce_Program_In_Python#What_we_want_to_do

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

mapreduce编程实例python-使用Python实现Hadoop MapReduce程序 的相关文章

  • 相机与imu的标定(Kalibr)

    在进行vio算法开发前最重要的事是对设备内参外参的标定 xff0c 其准确性直接决定了算法的有效性 xff0e 这里我将对最著名的kalibr标定工具的使用步骤进行说明 xff0c 包括安装 相机标定 imu标定 相机与imu联合标定等步骤

  • 解决cv_bridge依赖opencv版本问题

    1 问题来源 在安装ros的过程中 xff0c 系统会默认安装cv bridge库 xff0c 但该库指定了依赖的opencv库路径 xff0c 拿ros melodic版本来说 xff0c 默认依赖opencv库 usr lib x86

  • 使用ORB_SLAM3运行Realsense T265

    关于硬件 官网说明 使用说明 Realsense T265是一款跟踪相机 xff0c 配有两个FOV为111 7 x 108 6的广角相机 xff0c 并且带有IMU BMI055 惯性测量单元 设备内部配有vpu处理器并嵌入了建图和定位算

  • ceres-solver和g2o性能比较

    前言 ceres solver 和 g2o 是slam领域常见的优化器 xff0c 其中ceres solver被vins mono使用 xff0c 而g2o被orb slam3使用 xff0c 因此到底哪个优化器更适合于在slam算法开发

  • FreeRTOS的vTaskDelete使用说明

    FreeRTOS的vTaskDelete使用说明 函数说明 参数 xff1a xTaskToDelete 要删除的任务的任务句柄 返回值 无 说明 删除一个用函数xTaskCreate 或者xTaskCreateStatic 创建的任务 x

  • 机器学习——随机森林（Random Forest)

    1 随机森林 xff08 random forest xff09 简介 随机森林是一种集成算法 xff08 Ensemble Learning xff09 xff0c 它属于Bagging类型 xff0c 通过组合多个弱分类器 xff0c

  • 《基础知识——C和C++的主要区别》

    C和C 43 43 的主要区别 设计思想上 xff1a C 43 43 是面向对象的语言 xff0c 而C是面向过程的结构化编程语言 语法上 xff1a C 43 43 具有封装 继承和多态三种特性 C 43 43 相比C xff0c 增加

  • 数据库原理及应用(十三)E-R图、关系模式

    数据库设计的过程 数据分析 gt 数据建模 gt 关系数据库模式 gt 关系数据库管理 用户需求 gt 概念模型 E R Model gt 逻辑模型 xff08 三层结构 xff09 现实世界 gt 信息世界 gt 机器世界 概念设计工具E

  • Ubuntu数据备份与恢复工具(一)

    在我们日常工作中 xff0c 个人文件 业务数据及应用信息的备份与恢复策略是一个重要的环节 意外删除 硬件故障 操作失误 网络攻击 xff0c 甚至是自然灾害 xff0c 都可以直接或间接导不可估价的数据损失 为了避免损失 xff0c 缩少

  • 百度移动端面试回忆

    百度一面 xff1a 1 自我介绍 2 悲观锁和乐观锁 乐观锁 xff1a 总是认为不会产生并发问题 xff0c 每次去取数据的时候总认为不会有其他线程对数据进行修改 xff0c 因此不会上锁 xff0c 但是在更新时会判断其他线程在这之前

  • Quagga编译安装

    Quagga源码编译安装 1 Quagga下载 1 官网下载quagga 1 2 4 tar gz并拖入虚拟机桌面 2 解压到 opt目录下 sudo tar zxvf Desktop quagga 1 2 4 tar gz C opt 2

  • VINS-FUSION 源码 双目 单线程 按执行顺序阅读

    VINS FUSION 源码 双目 单线程 按执行顺序阅读 Keywords xff1a VINS FUSION vins 源码解读 源码梳理 vins数据结构 vinsfusion vins双目 双目vins 双目vinsfusion 双

  • 【C语言】__attribute__使用

    一 介绍 GNU C 的一大特色就是 attribute 机制attribute 可以设置函数属性 xff08 Function Attribute xff09 变量属性 xff08 Variable Attribute xff09 和类型

  • Ubuntu20.04下CUDA、cuDNN的详细安装与配置过程（图文）

    Ubuntu20 04下CUDA cuDNN的详细安装与配置过程 xff0c 亲测试可用 xff08 图文 xff09 一 NVIDIA xff08 英伟达 xff09 显卡驱动安装1 1 关闭系统自带驱动nouveau2 2 NVIDIA

  • 使用动量(Momentum)的SGD、使用Nesterov动量的SGD

    使用动量 Momentum 的SGD 使用Nesterov动量的SGD 参考 xff1a 使用动量 Momentum 的SGD 使用Nesterov动量的SGD 一 使用动量 Momentum 的随机梯度下降 虽然随机梯度下降是非常受欢迎的

  • Data Uncertainty Learning in Face Recognition

    Data Uncertainty Learning in Face Recognition 建模数据的不确定性对含噪音图像非常重要 xff0c 但对于人脸识别的研究却很少 先驱者的工作 35 通过将每个人脸图像嵌入建模为高斯分布来考虑不确定

  • ENAS代码解读

    ENAS代码解读 参考代码 xff1a https github com TDeVries enas pytorch 数据集 xff1a cifar10 main函数 xff1a span class token keyword def s

  • PC-DARTS Partial Channel Connections for Memory-Efficient Differentiable Architecture Search

    PC DARTS Partial Channel Connections for Memory Efficient Differentiable Architecture Search Abstract 可微体系结构搜索 xff08 DAR

  • deepsort代码解析

    DeepSort代码解析 项目地址 xff1a deepsort span class token keyword if span name span class token operator 61 61 span span class t

  • CBAM

    CBAM 我们提出了卷积块注意力模块 xff08 CBAM xff09 xff0c 这是一个简单而有效的前馈卷积神经网络的注意力模块 给定一个中间特征图 xff0c 我们的模块沿着通道和空间两个独立的维度依次推导注意力图 xff0c 然后将

随机推荐

热门标签