上海建工网站,天津商务网站建设,顺口好记的公司名字,华为手机网站建设策划书前言 分布式计算的精髓#xff0c;在于如何把抽象的计算流图#xff0c;转化为实实在在的分布式计算任务#xff0c;然后以并行计算的方式交付执行。今天这一讲#xff0c;我们就来聊一聊#xff0c;Spark 是如何实现分布式计算的。分布式计算的实现#xff0c;离不开两个…前言 分布式计算的精髓在于如何把抽象的计算流图转化为实实在在的分布式计算任务然后以并行计算的方式交付执行。今天这一讲我们就来聊一聊Spark 是如何实现分布式计算的。分布式计算的实现离不开两个关键要素一个是进程模型另一个是分布式的环境部署。接下来我们先去探讨 Spark 的进程模型然后再来介绍 Spark 都有哪些分布式部署方式。
触发计算流程图 函数 ##统计单词的次数import org.apache.spark.rdd.RDD// 这里的下划线_是占位符代表数据文件的根目录,hdfs的目录地址
val rootPath: String /user/hadoop/wikiOfSpark.txt
val file: String s${rootPath}// 读取文件内容
val lineRDD: RDD[String] spark.sparkContext.textFile(file)// 以行为单位做分词
val wordRDD: RDD[String] lineRDD.flatMap(line line.split( ))
val cleanWordRDD: RDD[String] wordRDD.filter(word !word.equals())// 把RDD元素转换为KeyValue的形式
val kvRDD: RDD[(String, Int)] cleanWordRDD.map(word (word, 1))
// 按照单词做分组计数
val wordCounts: RDD[(String, Int)] kvRDD.reduceByKey((x, y) x y)// 打印词频最高的5个词汇
wordCounts.map{case (k, v) (v, k)}.sortByKey(false).take(5)##########################
//统计相邻单词共现的次数假设我们再次改变 Word Count 的计算逻辑由原来统计单词的计数改为统计相邻单词共现的次数。import org.apache.spark.rdd.RDD// 这里的下划线_是占位符代表数据文件的根目录,hdfs的目录地址
val rootPath: String /user/hadoop/wikiOfSpark.txt
val file: String s${rootPath}// 读取文件内容
val lineRDD: RDD[String] spark.sparkContext.textFile(file)// 以行为单位提取相邻单词
val wordPairRDD: RDD[String] lineRDD.flatMap( line {// 将行转换为单词数组val words: Array[String] line.split( )// 将单个单词数组转换为相邻单词数组for (i - 0 until words.length - 1) yield words(i) - words(i1)
})val cleanWordRDD: RDD[String] wordPairRDD.filter(word !word.equals())// 把RDD元素转换为KeyValue的形式
val kvRDD: RDD[(String, Int)] cleanWordRDD.map(word (word, 1))
// 按照单词做分组计数
val wordCounts: RDD[(String, Int)] kvRDD.reduceByKey((x, y) x y)// 打印词频最高的5个词汇
wordCounts.map{case (k, v) (v, k)}.sortByKey(false).take(5)##对原来单词的计数改为对单词的哈希值计数在这种情况下。我们代码实现需要做哪些改动。import org.apache.spark.rdd.RDD
import java.security.MessageDigest// 这里的下划线_是占位符代表数据文件的根目录,hdfs的目录地址
val rootPath: String /user/hadoop/wikiOfSpark.txt
val file: String s${rootPath}// 读取文件内容
val lineRDD: RDD[String] spark.sparkContext.textFile(file)// 以行为单位做分词
val wordRDD: RDD[String] lineRDD.flatMap(line line.split( ))
val cleanWordRDD: RDD[String] wordRDD.filter(word !word.equals())
// 把普通RDD转换为Paired RDDval kvRDD: RDD[(String, Int)] cleanWordRDD.map{ word // 获取MD5对象实例val md5 MessageDigest.getInstance(MD5)// 使用MD5计算哈希值val hash md5.digest(word.getBytes).mkString// 返回哈希值与数字1的Pair(hash, 1)
}
// 按照单词做分组计数
val wordCounts: RDD[(String, Int)] kvRDD.reduceByKey((x, y) x y)// 打印词频最高的5个词汇
wordCounts.map{case (k, v) (v, k)}.sortByKey(false).take(5) import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
// 创建表
case class SiteViews(site_id: String, date: String, page_view: Int)
val siteViews Seq(SiteViews(a, 2021-05-20, 10),SiteViews(a, 2021-05-21, 11),SiteViews(a, 2021-05-22, 12),SiteViews(a, 2021-05-23, 12),SiteViews(a, 2021-05-24, 13),SiteViews(a, 2021-05-25, 14),SiteViews(a, 2021-05-26, 15),SiteViews(b, 2021-05-20, 21),SiteViews(b, 2021-05-21, 22),SiteViews(b, 2021-05-22, 22),SiteViews(b, 2021-05-23, 22),SiteViews(b, 2021-05-24, 23),SiteViews(b, 2021-05-25, 23),SiteViews(b, 2021-05-26, 25)
).toDF()
//
Window.partitionBy(column name|column)
// orderBy的语法
Window.orderBy(column name|column)
