【Day27-并列】大 大 大资料——操作巨量资料的必备观念MapReduce

在处理资料分析的时候，有的时候我们会需要将非常大量的资料之间进行一些交互的计算(例如矩阵乘法之类的)，而过程中会需要进行储存的记忆体空间需求就会变得很大，因此就会需要一种可以有效将本来需要大量暂存空间的计算方式给改良的计算过程，因此今天就介绍一下在做这块的部分吧

MapReduce

Mapreduce在巨量资料处理中的概念是利用了将可以把原本都要一起计算的东西，拆分成许多小的计算步骤，而每个可并行的小步骤之间又可以独立计算，因此可以减少记忆体空间的需求或是可以用多台资源进行计算。

Map——将每个步骤的最小元素进行转换

我们这边以计算单字数量为例，那如果要按照传统的计算方式我们可能会需要先将所有的内容存到同一个地方才能计算，那这边我们假设这边的步骤是可以分散进行的，那计算单字数量一共可以拆分成两步

• 第一步：将所有单字都计数1
• 第二步：将相同单字(key)的每两个元素的计数相加，结合成一个元素

words = ["Apple", "Banana", "Watermelon", "Banana", "Apple", "Apple"]
words_mapped = list(map(lambda x: (x,1), words))
words_mapped

输出：

[('Apple', 1),
 ('Banana', 1),
 ('Watermelon', 1),
 ('Banana', 1),
 ('Apple', 1),
 ('Apple', 1)]

Reduce——前一步骤的结果整合

这边先以pandas作为演练

import pandas as pd
df = pd.DataFrame(words_mapped, columns = ["word","count"])
df

输出：

df.groupby("word").agg({"sum"})

输出：

以大矩阵乘法为例示范pyspark

那实际上通常在跑巨量资料的一个架构是spark，而它也有在python上的套件

lines = sc.textFile("inputdata.txt")



# %%
def mapper_input(line):
    matrix_list = line.split(",")
    matrix_map = (matrix_list[0], [matrix_list[1],matrix_list[2],matrix_list[3]])
    return matrix_map

linesRDD = lines.map(mapper_input)

# 因为有了`'M'`或是`'N'`的key值，所以可以将资料分成M矩阵和N矩阵  
# 这边用RDD的`filter`来进行
# %%
M_RDD = linesRDD.filter(lambda x : "M" in x[0]) # [0]is the key value
N_RDD = linesRDD.filter(lambda x : "N" in x[0])

# Multiply


# %%
M_RDD_matrix = M_RDD.map(lambda x : (x[1][1], [x[0],x[1][0],x[1][2]] ))
N_RDD_matrix = N_RDD.map(lambda x : (x[1][0], [x[0],x[1][1],x[1][2]] ))

MN_RDD = M_RDD_matrix.join(N_RDD_matrix)


# %%
P_RDD_matrix = MN_RDD.map(lambda x: ((int(x[1][0][1]),int(x[1][1][1])),int(x[1][0][2])*int(x[1][1][2])) )

P_RDD = P_RDD_matrix.reduceByKey(lambda x,y: int(x)+int(y))

# 为了符合输出个格式，先用`sortByKey()`来排序

# %%
P_RDD_sort = P_RDD.sortByKey()

# %%
P_RDD_sort.collect()

# %% [markdown]
# 将排序好的RDD重新用`map()`来把$((i,k),P_{ij})$映射到同一层以方便输出

# %%
P_RDD_reshape = P_RDD_sort.map(lambda x : (x[0][0],x[0][1],x[1]) )

# %%
output = P_RDD_reshape.collect()

产生基本RDD元素

定义mapper_input()function，用来将原本每一行的资料切分和整理
输入格式： M,0,0,10
输出格式： ('M', ['0', '0', '10']),

get martix M and N by filter

因为有了'M'或是'N'的key值，所以可以将资料分成M矩阵和N矩阵
这边用RDD的filter来进行

相乘

因为,两个矩阵可以相乘的话一定要有共同的j，所以将M和N分别以row vector和column vector来表示以便进行後续的计算

在进行大矩阵运算的时候的重点在於找出可以不会互相影响的步骤而拆开进行

输入输出

输入

每行表示M或N矩阵的第i,j个元素为多少

输出

每行表示M或N矩阵的第i,j个元素为多少