当前位置:   article > 正文

Spark RDD编程基础_rdd.map

rdd.map

一、数据读写

(1)从文件系统加载数据创建RDD

  ①本地文件:sc.textFile("file:///data/spark/buyer_favorite")

  ②HDFS文件:sc.textFile("hdfs://localhost:9000/spark/buyer_favorite")

(2)通过并行集合创建RDD

  val array = Array(1,2,3,4,5)

  val rdd = sc.parallelize(array)

(3)将RDD写入到文本文件

   使用saveAsTextFile()函数,要求提供一个不存在的目录名称

  *使用repartition重新设置分区个数

二、RDD常见操作

(1)转换操作

①  filter(func):筛选出满足函数func的元素,并返回一个新的数据集。

②  map(func):将每一个元素传递到函数func中,并将结果返回为一个新的数据集。

③  flatMap(func):类似map(),但是每个输入都可以映射0到多个输出结果。可以理解为:

    第一步:执行map()

    第二步:把map操作得到的数组中每个元素“拍扁”(flat)

④  groupByKey():对具有相同键的值进行分组,返回一个(key,Iterable)形式数据集。

⑤  reduceByKey(func):先执行groupByKey()操作得到(key,value-list),根据func对value-list进行操作。

(2)行动操作

①  count():返回数据集中元素个数。

②  collect():以数组的形式返回数据集中的所有元素。

③  first():返回数据集第一个元素。

④  take(n):以数组的形式返回数据集中的前n个元素。

⑤  reduce(func):通过函数func(输入两个参数,返回一个值)聚合数据集中的元素。

⑥  foreach(func):将数据集中每个元素传递到func中运行。

三、键值对RDD

(1)创建键值对RDD

①val lines = sc.textTextFile(file:///usr/local/word.txt)

  var pairRDD=lines.flatMap(_.split("\t")).map(word=>(word,1))

②var pairRDD=rdd.map(word=>(word,1))

(2)常用键值对RDD操作

reduceByKey(func)

groupByKey()

keys:返回一个新的RDD.

values

sortByKey()

sortBy()

mapValues(func)

join():内连接,对于给定的两个数据集(k,v1)和(k,v2),当两个数据集都存在key才会被输出,最注重得到一个(k,(v1,v2))类型的数据集。

四、几个简单例子

(1)

WordCount统计:某电商网站记录了大量的用户对商品的收藏数据,并将数据存储在名为buyer_favorite的文本文件中。文本数据格式如下:
用户id(buyer_id),商品id(goods_id),收藏日期(dt);现要求统计用户收藏数据中,每个用户收藏商品数量。

rdd.map(line=> (line.split('\t')(0),1)).reduceByKey(_+_).collect

(2)

去重:使用spark-shell,对上述实验中,用户收藏数据文件进行统计。根据商品ID进行去重,统计用户收藏数据中都有哪些商品被收藏。

rdd.map(line => line.split('\t')(1)).distinct.collect      相当于
rdd.map(line=>(line.split('\t')),1).reduceByKey(_+_).map(_._1).collect

(3)排序:电商网站都会对商品的访问情况进行统计,现有一个goods_visit文件,存储了电商网站中的各种商品以及此各个商品的点击次数。

商品id(goods_id) 点击次数(click_num)现根据商品的点击次数进行排序,并输出所有商品。

rdd1.map(line => ( line.split('\t')(1).toInt, line.split('\t')(0) ) ).sortByKey(true).map(_._2).collect

(4)Join:现有某电商在2011年12月15日的部分交易数据。数据有订单表orders和订单明细表order_items,表结构及数据分别为:

orders表:(订单id order_id, 订单号 order_number, 买家ID buyer_id, 下单日期 create_dt)
order_items表:(明细ID item_id, 订单ID order_id, 商品ID goods_id )
orders表和order_items表,通过订单id进行关联,是一对多的关系。

下面开启spark-shell,查询在当天该电商网站,都有哪些用户购买了什么商品

rdd1.map(line=> (line.split('\t')(0), line.split('\t')(2)) ) 
    join
rdd2.map(line=> (line.split('\t')(1), line.split('\t')(2)) )
     .collect

(5)求平均值:电商网站都会对商品的访问情况进行统计。现有一个goods_visit文件,存储了全部商品及各商品的点击次数。还有一个文件goods,记录了商品的基本信息。两张表的数据结构如下:

goods表:商品ID(goods_id),商品状态(goods_status),商品分类id(cat_id),评分(goods_score)

goods_visit表:商品ID(goods_id),商品点击次数(click_num)

商品表(goods)及商品访问情况表(goods_visit)可以根据商品id进行关联。现在统计每个分类下,商品的平均点击次数是多少?

val rdd11 = rdd1.map(line=> (line.split('\t')(0), line.split('\t')(2)) )  
val rdd22 = rdd2.map(line=> (line.split('\t')(0), line.split('\t')(1)) )  
val rddjoin = rdd11 join rdd22  
rddjoin.map(x=>{(x._2._1, (x._2._2.toLong, 1))}).reduceByKey((x,y)=>{(x._1+y._1, x._2+y._2)}).map(x=>  
{(x._1, x._2._1*1.0/x._2._2)}).collect

 

声明:本文内容由网友自发贡献,不代表【wpsshop博客】立场,版权归原作者所有,本站不承担相应法律责任。如您发现有侵权的内容,请联系我们。转载请注明出处:https://www.wpsshop.cn/w/花生_TL007/article/detail/600846
推荐阅读
相关标签
  

闽ICP备14008679号