mapreduce数据分析

89 阅读 0 评论 59 点赞

我是靠谱客的博主冷酷眼神，最近开发中收集的这篇文章主要介绍mapreduce数据分析，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

spark多了有向无环图，DAG，计算两节点之间的最优路径

可以使用的语言：java,python,scala,r

核心：RDD

hdfs:是文件系统，通过sc...,转成rdd，hadoop distributed file system

两种类型：并行集合和文件系统 sc.parallelize() sc.textFile() 得到rdd

两种计算方式：变换（transformation）例如：map、filter、flatmap等，是并行化的，几台服务器，不同的datanode,worker上执行,Excutor。

操作（actions）例如：count、collect、take、reduce等，driver上执行，sparkcontext

mapreduce必须编程完成后才能统一提交执行,只能使用java语言。

hadoop mapreduce数据分析：基于java语言，压缩包都是jar包

map打散格式 reduce统计

cd opt-->cd hadoop-2.9.0-->cd share-->cd hadoop-->cd mapreduce

在mapreduce下：

hadoop jar hadoop-mapreduce-examples-2.7.5jar wordcount /user/hadoop/haha /user/output

数据分析一直就是mapreduce,wordcount

查看结果：hadoop fs -cat /user/output/part-r-00000

scala语言数据分析：语法简洁，可以写在一行，是基于java的，看菜鸟教程

命令：spark-shell

var lines=sc.textFile(”hdfs://python:9000/user/hadoop/data1.txt")

lines.collect()查看

lines.map（x=>x.split(' ')).flatMap(x=>for(i<-x) yield(i,1).groupByKey().map(x=>(x._1,x._2.sum)).collect

#map里x指的是每一个元素，拆分成三个数组，返回新的rdd，flatMap里x是每个数组，进行迭代，生成此格式，平铺展开成一个数组，如果用map还是三个数组，groupByKey根据键进行分组，reduceByKey根据键进行统计总数，都是键值对形式，x._1代表第一个元素，第二个元素求和

第二种：

lines.map（x=>x.split(' ')).flatMap(x=>for(i<-x) yield(i,1).reduceByKey((x,y)=>x+y).collect

#=>相当于lambda

python语言数据分析：python函数库丰富，效率低

命令：pyspark

def a(x):

return (x/2)

hah=sc.parallilize([23,11,23])

dd=hah.map(a).collect()#装饰，这句话就是调用，不需要加括号。

([1,2,3,2,3,2,3)]

rdd.conntByVlaue()#直接返回键值对形式{1:1,2:3,3:3}

wget下载网站

python5:8088 hadoop操作日志

[(('r1', 'aa'), 23), (('r2', 'aab'), 25), (('r1', 'adc'), 29), (('r2', 'aq'), 21), (('r3', 'aw'), 23), (('r1', 'aa'), 53), (('r2', 'aww'), 54), (('r3', 'qq'), 45), (('r3', 'qq'), 22), (('r1', 'aa'), 12)]

rs=rdd4.collect()#转成列表形式，见上

class CompareValue(object):
def __init__(self,obj):#初始化
self.obj=obj
def __lt__(self,other):#重写小于方法
return self.obj>other.obj#自定义，写大于、小于都行
rs=sorted(rs,key=(lambda x:(x[0][0],CompareValue(x[0][1]))),reverse=False)

print(rs)

flatMapValues操作：在pyspark中找,都有,将值打散（API函数: application programming interface应用程序接口)

x = sc.parallelize([("a", ["x", "y", "z"]), ("b", ["p", "r"])])
def f(x):
return x
print(x.flatMapValues(f).collect())

#[('a', 'x'), ('a', 'y'), ('a', 'z'), ('b', 'p'), ('b', 'r')]