本文是在学习 “7个实例全面掌握Hadoop MapReduce” 这篇文章后&＃xff0c;用Spark重新实现了里面的例子&＃xff0c;希望对初学Spark的童学有帮助&＃xff0c;欢迎大牛们留言提意见。
上篇包含4个实例&＃xff1a;
- 1&＃xff1a;自定义对象序列化
- 2&＃xff1a;自定义分区
- 3&＃xff1a;计算出每组订单中金额最大的记录
- 4&＃xff1a;合并多个小文件
下篇包含3个实例&＃xff1a;
- 5&＃xff1a;分组输出到多个文件
- 6&＃xff1a;join操作
- 7&＃xff1a;计算出用户间的共同好友
实例1&＃xff1a;自定义对象序列化
1&＃xff09;需求
需要统计手机用户流量日志&＃xff0c;日志内容实例&＃xff1a;
要把同一个用户的上行流量、下行流量进行累加&＃xff0c;并计算出综合。
例如上面的13897230503有两条记录&＃xff0c;就要对这两条记录进行累加&＃xff0c;计算总和&＃xff0c;得到&＃xff1a;
13897230503&＃xff0c;500&＃xff0c;1600&＃xff0c;2100
2&＃xff09;代码实践
package mapReduceimport org.apache.spark.sql.{Dataset, SparkSession}object FlowCount {def main(args: Array[String]): Unit &＃61; {val spark &＃61; SparkSession.builder().appName("FlowCount").master("local[1]").getOrCreate()import spark.implicits._import org.apache.spark.sql.functions._// Optional, help avoid errors due to long lineagespark.sparkContext.setLogLevel("WARN")spark.sparkContext.setCheckpointDir("hdfs://127.0.0.1:9090/tmp/")val base &＃61; "hdfs://127.0.0.1:9090/user/ds/"val lines &＃61; spark.read.textFile(base &＃43; "flowcount.txt")/* 生成DataSet */val flowData &＃61; lines.map(preparation).toDF("phone", "upFlow", "dFlow", "sumFlow")flowData.show(5)/* 以手机号为key&＃xff0c;对数值求和 */val sumData &＃61; flowData.groupBy("phone").agg(sum("upFlow"),sum("dFlow"), sum("sumFlow"))sumData.show(5)spark.stop()}def preparation(str: String): FlowBean &＃61; {//切分字段val fields &＃61; str.split(" ")//assert(fields.size &＃61;&＃61; 4)//组装对象var phone &＃61; fields(0).toString.trimvar upFlow &＃61; fields(1).toLongvar dFlow &＃61; fields(2).toLongFlowBean(phone, upFlow, dFlow, upFlow &＃43; dFlow)}}/** phone:手机号,* upFlow:上行流量, dFlow:下行流量, sumFlow:流量合计*/
case class FlowBean(phone:String, upFlow:Long, dFlow:Long, sumFlow:Long)
3&＃xff09;运行结果
实例2&＃xff1a;自定义分区
1&＃xff09;需求
还是以上个例子的手机用户流量日志为例&＃xff0c;
在上个例子的统计需要基础上添加一个新需求&＃xff1a;按省份统计&＃xff0c;不同省份的手机号放到不同的文件里。
例如137表示属于河北&＃xff0c;138属于河南&＃xff0c;那么在结果输出时&＃xff0c;他们分别在不同的文件中。
2&＃xff09;代码实践
package mapReduceimport org.apache.spark.sql.{Dataset, SparkSession}object FlowPartitioner {def main(args: Array[String]): Unit &＃61; {val spark &＃61; SparkSession.builder().appName("FlowCount").master("local[1]").getOrCreate()import spark.implicits._import org.apache.spark.sql.functions._// Optional, help avoid errors due to long lineagespark.sparkContext.setLogLevel("WARN")spark.sparkContext.setCheckpointDir("hdfs://127.0.0.1:9090/tmp/")val base &＃61; "hdfs://127.0.0.1:9090/user/ds/"val lines &＃61; spark.read.textFile(base &＃43; "flowcount.txt")/* 指定数据分区器 */val lineData &＃61; lines.map(line &＃61;> (line.split(" ")(0), line)).rdd.partitionBy(new ProvincePartitioner(3))/* 生成DataSet, 持久化避免重复混洗 */val flowData &＃61; lineData.values.map(preparation).toDF("phone", "upFlow", "dFlow", "sumFlow").persist()flowData.show(5)spark.stop()}def preparation(str: String): FlowBean02 &＃61; {//切分字段val fields &＃61; str.split(" ")//assert(fields.size &＃61;&＃61; 4)//组装对象var phone &＃61; fields(0).toString.trimvar upFlow &＃61; fields(1).toLongvar dFlow &＃61; fields(2).toLongFlowBean02(phone, upFlow, dFlow, upFlow &＃43; dFlow)}}/** phone:手机号,* upFlow:上行流量, dFlow:下行流量, sumFlow:流量合计*/
case class FlowBean02(phone:String, upFlow:Long, dFlow:Long, sumFlow:Long)import org.apache.spark.Partitioner/** 自定义partitioner* 根据手机号前缀分区*/
class ProvincePartitioner(val num: Int) extends Partitioner {override def numPartitions: Int &＃61; numoverride def getPartition(key: Any): Int &＃61; {//手机号前3位val prefix &＃61; key.toString().substring(0, 3)//用分区个数取模prefix.toInt % num}
}
3&＃xff09;运行结果
实例3&＃xff1a;计算出每组订单中金额最大的记录
1&＃xff09;需求
有如下订单数据&＃xff1a;
求出每一个订单中成交金额最大的一笔交易。
2&＃xff09;代码实践
package mapReduceimport mapReduce.FlowPartitioner.preparation
import org.apache.spark.sql.SparkSessionobject GroupSort {def main(args: Array[String]): Unit &＃61; {val spark &＃61; SparkSession.builder().appName("FlowCount").master("local[1]").getOrCreate()import spark.implicits._import org.apache.spark.sql.functions._// Optional, help avoid errors due to long lineagespark.sparkContext.setLogLevel("WARN")spark.sparkContext.setCheckpointDir("hdfs://127.0.0.1:9090/tmp/")val base &＃61; "hdfs://127.0.0.1:9090/user/ds/"val lines &＃61; spark.read.textFile(base &＃43; "groupsort.txt")/* 默认分区器是hashcode */val lineData &＃61; lines.map(preparation).repartition(3).toDF("itemid", "goodsid", "amount")val sortData &＃61; lineData.sortWithinPartitions("itemid", "amount")sortData.show(10)/* 取出金额最大的订单 */val maxData &＃61; sortData.groupBy("itemid").agg(max("amount")).sort($"itemid".asc)maxData.show(10)spark.stop()}def preparation(str: String): ItemBean &＃61; {//切分字段val fields &＃61; str.split(" ")//assert(fields.size &＃61;&＃61; 4)//组装对象var itemid &＃61; fields(0).toString.trimvar goodsid &＃61; fields(1).toString.trimvar amount &＃61; fields(2).toDoubleItemBean(itemid, goodsid, amount)}}case class ItemBean(itemid:String, goodsid:String, amount:Double)
3&＃xff09;运行结果
实例4&＃xff1a;合并多个小文件
1&＃xff09;需求
目标文件中有大量的小文件&＃xff0c;会造成分配任务和资源的开销比实际的计算开销还打&＃xff0c;这就产生了效率损耗。
需要先把一些小文件合并成一个大文件。
2&＃xff09;代码实践
package mapReduceimport org.apache.spark.sql.SparkSessionobject ManyToOne {def main(args: Array[String]): Unit &＃61; {val spark &＃61; SparkSession.builder().appName("FlowCount").master("local[1]").getOrCreate()// Optional, help avoid errors due to long lineagespark.sparkContext.setLogLevel("WARN")spark.sparkContext.setCheckpointDir("hdfs://127.0.0.1:9090/tmp/")/* 1.可以将path里的所有文件内容读出 *///spark.read.textFile(path)/* 2.指定文件列表 */val base &＃61; "hdfs://127.0.0.1:9090/user/ds/"val fileList &＃61; Array(base &＃43; "groupsort.txt", base &＃43; "flowcount.txt")/* union 连接 */val fileRDD &＃61; fileList.map(spark.read.textFile(_))val lineData &＃61; fileRDD.reduce((x,y)&＃61;> x.union(y))lineData.show(20)spark.stop()}}
3&＃xff09;运行结果
如果数据很大&＃xff0c;难以在单机内存上装下&＃xff0c;以上操作可能会造成单机内存不足。
转自 "如何使saveAsTextFile只生成一个文件"
在RDD上调用 coalesce(1,true).saveAsTextFile()&＃xff0c;
或者&＃xff0c;可以调用 repartition(1)&＃xff0c;它其实是coalesce的一个包装&＃xff0c;默认第二个参数为true。
但代价是巨大的&＃xff0c;因为Spark面对的是大量的数据&＃xff0c;并且是并行执行的&＃xff0c;如果强行要求最后只有一个分区&＃xff0c;必然导致大量的磁盘IO和网络IO产生。
安全的做法&＃xff0c;可以采用HDFS磁盘合并操作&＃xff1a;
hadoop fs -cat /hdfs/result/part-* > /data/test.log
或
hadoop fs -getmerge /hdfs/result /data/test.log
京公网安备 11010802041100号