设为首页 加入收藏

TOP

第一个Hadoop程序
2018-12-07 00:41:04 】 浏览:6
Tags:一个 Hadoop 程序
版权声明:转载请注明出处 https://blog.csdn.net/csdn_lzw/article/details/80820721

需求:有100个文件(每个大概10G,300万个样例)每个样例可以得到对应的类别属性属性值。统计属性值出现的次数
类似 wordcount
其中 word 是 类(cat1-cat3)属性属性值

mapper.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys
reload(sys)
sys.setdefaultencoding("utf-8")
import  json
import os
for line in sys.stdin:
    s = line.split('\t')[1]
    obj = json.loads(s)

    cat = ''
    cat = obj['l_cat'][0] + '-' + obj['l_cat'][2]
    word = ''
    word = cat + ',' + 'cat_3' +','+cat
    print '%s\t%s' % (word, 1)

    word = ''
    word = cat + ','+'brand'+','+obj['brand']
    print '%s\t%s' % (word, 1)

    for att, val in obj['d_attr'].items():
        if att =='商品毛重' or att == '商品名称' or att == '商品编号' \
           or att == '货号' or att == '店铺' or len(val)>10:
               continue
        word = ''
        word = cat+','+att+','+val
        print '%s\t%s' % (word, 1)

reducer.py

#!/usr/bin/env python
# -*- coding: utf-8 -*- 
import sys
current_word = None
current_count = 0
word = None
for line in sys.stdin:
    line = line.strip()
    word, count = line.split('\t', 1)
    try:
        count = int(count)
    except ValueError:
        continue
    if current_word == word:
        current_count += count
    else:
        if current_word:
            print '%s\t%s' % (current_word, current_count)
        current_count = count
        current_word = word
if current_word == word:
    print '%s\t%s' % (current_word, current_count)

本地测试 part-00000-sample是输入 ,输出到out

cat part-00000-sample | python mapper.py | sort -k 1 | python reducer.py>out

查看测试的结果

cat out|less

这里写图片描述

本地测试成功了才可以提交到Hadoop 上
通过脚本提交到Hadoop 上

!!!若Hadoop上python 版本低,需将python 环境打包到Hadoop上

-cacheArchive "/app/ssg/nlp/uu/linzhiwei02/tools/python27.tar.gz#python27"\

参考文献 https://zhuanlan.zhihu.com/p/34903460

脚本 run_hadoop.sh

#!/bin/bash
hadoop="/home/linzhiwei02/tools/hadoop-client-heng/hadoop/bin/hadoop"
$hadoop streaming \
    -D mapred.job.queue.name="nlp"\
    -D mapred.job.name="nlp-linzhiwei02-count-item"\
    -D mapred.job.priority=NORMAL \
    -D mapred.map.tasks=400 \
    -D mapred.reduce.tasks=100\
    -cacheArchive "/app/ssg/nlp/uu/linzhiwei02/tools/python27.tar.gz#python27"\
    -input "/app/ssg/nlp/uu/linzhiwei02/commodity_title/20180315_label" \
    -output "/app/ssg/nlp/uu/linzhiwei02/Output5" \
    -file ./mapper.py \
    -file ./reducer.py\
    -mapper "python27/bin/python  mapper.py" \
    -reducer "python27/bin/python reducer.py" \
    -partitioner "org.apache.hadoop.mapred.lib.KeyFieldBasedPartitioner"

注意反斜杠之后回车,不能有多余的空格
解压之后的python 的路径要添加到 -mapper 和 -reducer 中


编程开发网
】【打印繁体】【投稿】【收藏】 【推荐】【举报】【评论】 【关闭】 【返回顶部
上一篇结合案例讲解MapReduce重要知识点.. 下一篇何时不用Hadoop?

评论

帐  号: 密码: (新用户注册)
验 证 码:
表  情:
内  容:

array(4) { ["type"]=> int(8) ["message"]=> string(24) "Undefined variable: jobs" ["file"]=> string(32) "/mnt/wp/cppentry/do/bencandy.php" ["line"]=> int(214) }