word_count实践操作

快来打我* 2022-05-15 14:07 156阅读 0赞

最基础的Work\_count：

Map.py

import sys

import time

import re

p = re.compile(r'\\w+')

for line in sys.stdin:

ss = line.strip().split(' ')

for s in ss:

if len(p.findall(s)) < 1:

continue

word = p.findall(s)\[0\].lower()

\#word = s.strip()

if word.strip() != "":

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

Red.py

\#!/usr/bin/python

import sys

current\_word = None

sum = 0

for line in sys.stdin:

word, val = line.strip().split('\\t')

if current\_word == None:

current\_word = word

if current\_word != word:

print "%s\\t%s" % (current\_word, sum)

current\_word = word

sum = 0

sum += int(val)

print "%s\\t%s" % (current\_word, str(sum))

执行：

cat The\_Man\_of\_Property.txt |python map.py |sort -k1|python red.py

白名单word\_count：

Map.py:

import sys

import time

def read\_local\_file\_func(f):

word\_set = set()

file\_in = open(f, 'r')

for line in file\_in:

word = line.strip()

word\_set.add(word)

return word\_set

def mapper\_func(white\_list\_fd):

word\_set = read\_local\_file\_func(white\_list\_fd)

for line in sys.stdin:

ss = line.strip().split(' ')

for s in ss:

word = s.strip()

if word != "" and (word in word\_set):

\#print s + "\\t" + "1"

\#print '\\t'.join(\[s, "1"\])

print "%s\\t%s" % (s, 1)

if \_\_name\_\_ == "\_\_main\_\_":

module = sys.modules\[\_\_name\_\_\]

func = getattr(module, sys.argv\[1\])

args = None

if len(sys.argv) > 1:

args = sys.argv\[2:\]

func(\*args)

Red.py：和上一样

执行：

cat The\_Man\_of\_Property.txt |python map2.py mapper\_func white\_list |sort -k1 |python red.py

两个文件，key有相同的，value不相同，最终将两个文件合并（相同的key，把它的价格，和商品名字拿出来），如何解决？

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70][]

执行

Map\_a.py

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    # coding=UTF-8
    
    import sys
    for line in sys.stdin:
        ss = line.strip().split('  ')
        key = ss[0]
        value = ss[1]
        print "%s\t1\t%s" % (key, value)

map\_b.py：

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    # coding=UTF-8
    
    import sys
    for line in sys.stdin:
        ss = line.strip().split('   ')
        key = ss[0]
        value = ss[1]
        print "%s\t2\t%s" % (key, value)

执行 cat a\_join.txt |python map\_a.py >a1

执行 cat b\_join.txt |python map\_b.py >b2

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 1][]

编写map\_join.py

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    # coding=UTF-8
    
    import sys
    
    val_1 = ''
    for line in sys.stdin:
        key, flag, val = line.strip().split('\t')
    
        if flag == '1':
            val_1 = val
        elif flag == '2' and val_1 != '':
            val_2 = val
            print"%s\t%s\t%s" %(key, val_1, val_2)
            val_1 = ''

执行cat a1 b2 |sort -k1 |python map\_join.py 完成map操作

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 2][]

hadoop上运行的sh脚本：

set -e -x

HADOOP\_CMD="/usr/local/src/hadoop-2.7.3/bin/hadoop"  
STREAM\_JAR\_PATH="/usr/local/src/hadoop-2.7.3/share/hadoop/tools/lib/hadoop-streaming-2.7.3.jar"

INPUT\_FILE\_PATH\_A="/data/a\_join.txt"  
INPUT\_FILE\_PATH\_B="/data/b\_join.txt"

OUTPUT\_A\_PATH="/output/a"  
OUTPUT\_B\_PATH="/output/b"

OUTPUT\_JOIN\_PATH="/output/join"

$HADOOP\_CMD fs -rmr -skipTrash $OUTPUT\_A\_PATH $OUTPUT\_B\_PATH $OUTPUT\_JOIN\_PATH  
\#$HADOOP\_CMD fs -rmr -skipTrash $OUTPUT\_JOIN\_PATH

\# Step 1.  
$HADOOP\_CMD jar $STREAM\_JAR\_PATH \\  
    -input $INPUT\_FILE\_PATH\_A \\  
    -output $OUTPUT\_A\_PATH \\  
    -mapper "python map\_a.py" \\  
    -file ./map\_a.py \\

\# Step 2.  
$HADOOP\_CMD jar $STREAM\_JAR\_PATH \\  
    -input $INPUT\_FILE\_PATH\_B \\  
    -output $OUTPUT\_B\_PATH \\  
    -mapper "python map\_b.py" \\  
    -file ./map\_b.py \\

\# Step 3.  
$HADOOP\_CMD jar $STREAM\_JAR\_PATH \\  
    -input $OUTPUT\_A\_PATH,$OUTPUT\_B\_PATH \\  
    -output $OUTPUT\_JOIN\_PATH \\  
    -mapper "cat" \\  
    -reducer "python red\_join.py" \\  
    -file ./red\_join.py \\  
    -jobconf stream.num.map.output.key.fields=2 \\  
    -jobconf num.key.fields.for.partition=1

运行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 3][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70]: /images/20220503/65a26b97c9ca44fb927c4e25bdd8771d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 1]: /images/20220503/08ccf4e5c1c34bd2b7ac7105d9c5f940.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 2]: /images/20220503/20989dec209344c1a7b673e1fb9a267f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzI4Mjg2MDI3_size_16_color_FFFFFF_t_70 3]: /images/20220503/f93d267181ca46e48e78d76e5421f98e.png