Join实现-MRJob-Python数据分析(16)

更新：2022-05-23 13:17:04编辑：Story归类：数据库人气：202

1.1. 前言

前面我们解析出了CDN IP 和用户真实IP 的访问次数，但是我们要如何知道该IP是来自哪个地区呢？这时候我们就需要用到在area_ip.csv 中的数据了。

这时候就需要使用 M/R 来实现 IP 和 area_ip 的 join 行为。使用 M/R 来实现join其实就是笛卡尔积之后的过滤。如果数据量大的话将会是一个十分耗时的过程。而且我们 area_ip.csv 的数据是一个IP范围的数据，我们还需要将它展开，如：

start_ip: 0.0.0.0 stop_ip: 0.0.0.3

转化为四条数据

0.0.0.0
0.0.0.1
0.0.0.2
0.0.0.3

0.0.0.0

0.0.0.1

0.0.0.2

0.0.0.3

初步一算原来 30M 多的 area_ip 数据就会变成 30G 多有好几十亿的数据这个用 M/R 做起join起来那可想而知了。

为了演示 M/R 的join 我们这边使用一个比较简单的数据来展示一下

1.2. 需求

求好友间接关系

1.3. 数据

表1
    A B
    B C
    C D
    D E
    D A
    E F
    E B
    F G

表 1

A B

B C

C D

D E

D A

E F

E B

F G

1.4. Mapper 步骤

键值互换变成两表关联

表1    表2
A B    B A
B C    C B
C D    D C
D E    E D
D A    A D
E F    F E
E B    B E
F G    G F

表 1 表 2

A B B A

B C C B

C D D C

D E E D

D A A D

E F F E

E B B E

F G G F

1.5. Shuffle 步骤

拆分成两半并打上标记

A: B
B: A
B: C
C: B
C: D
D: C
D: E
E: D
D: A
A: D
E: F
F: E
E: B
B: E
F: G
G: F

A : B

B : A

B : C

C : B

C : D

D : C

D : E

E : D

D : A

A : D

E : F

F : E

E : B

B : E

F : G

G : F

1.6. 获取key相同的集合

A: B,D
B: A,C,E
C: B,D
D: C,E,A
E: D,F,B
F: E,G

A : B , D

B : A , C , E

C : B , D

D : C , E , A

E : D , F , B

F : E , G

1.7. Reduce 步骤

取出每个key的值进行笛卡尔集从而得出有间接好友关系

原本要去重的这边就不去了

B D
D B
A C
A E
C A
C E
E A
E C
B D
D B
A C
A E
C A
C E
E A
E C
B D
B F
D B
D F
F B
F D
E G
G E

B D

D B

A C

A E

C A

C E

E A

E C

B D

D B

A C

A E

C A

C E

E A

E C

B D

B F

D B

D F

F B

F D

E G

G E

1.8. 程序代码

vim mr_friendship.py
 
#!/usr/bin/env <a href="http://www.ttlsa.com/python/" title="Python"target="_blank">python</a>
# -*- coding: utf-8 -*-
 
from mrjob.job import MRJob
 
class MRFriendship(MRJob):
 
    def mapper(self, _, line):
        # 交换好友 key和value 分别输出
        f1, f2 = line.split()
        yield f1, f2
        yield f2, f1
 
    def reducer(self, key, values):
        # 对通过了shuffle(partition, sort, combine)的数据进行笛卡尔集
        friend_keys = values # 用于key输出
        friend_values = values # 用于value输出
 
        for friend_key in friend_keys: # 第一层循环作为key
            for friend_value in friend_values: # 第二次循环作为value
                # 组合 key 和 value 并输出
 
def main():
    MRFriendship.run()
 
if __name__ == '__main__':
    main()

vim mr_friendship . py

#!/usr/bin/env python

# -*- coding: utf-8 -*-

from mrjob . job import MRJob

class MRFriendship ( MRJob ) :

def mapper ( self , _ , line ) :

# 交换好友 key和value 分别输出

f1 , f2 = line . split ( )

yield f1 , f2

yield f2 , f1

def reducer ( self , key , values ) :

# 对通过了shuffle(partition, sort, combine)的数据进行笛卡尔集

friend_keys = values # 用于key输出

friend_values = values # 用于value输出

for friend_key in friend_keys : # 第一层循环作为key

for friend_value in friend_values : # 第二次循环作为value

# 组合 key 和 value 并输出

def main ( ) :

MRFriendship . run ( )

if __name__ == '__main__' :

main ( )

1.9. 准备数据

vim friendship.txt
A B
B C
C D
D E
D A
E F
E B
F G

vim friendship . txt

A B

B C

C D

D E

D A

E F

E B

F G

1.10. 执行程序

python mr_friendship.py < friendship.txt
 
"B"     "D"
"D"     "B"
"A"     "C"
"A"     "E"
"C"     "A"
"C"     "E"
"E"     "A"
"E"     "C"
"B"     "D"
"D"     "B"
"A"     "C"
"A"     "E"
"C"     "A"
"C"     "E"
"E"     "A"
"E"     "C"
"B"     "D"
"B"     "F"
"D"     "B"
"D"     "F"
"F"     "B"
"F"     "D"
"E"     "G"
"G"     "E"