1.利用scala写单词统计
package com.dt.spark
import org.apache.spark.{SparkConf, SparkContext}
object WordCount {
def main(args: Array[String]) {
val conf = new SparkConf().setAppName("WordCount").setMaster("local") //单机
// val conf = new SparkConf().setAppName("WordCount").setMaster("spark://master:7077") // 集群
val sc = new SparkContext(conf)
// val lines = sc.textFile("F:\\scala\\WordCount\\src\\com\\dt\\spark\\file.txt") // 本地文件
// val lines = sc.textFile("/library/wordcount/input") // HDFS文件
val lines = sc.textFile("hdfs://10.65.123.102:9000/user/hive/warehouse/applog/ns_date=20180622/hash=A481-CEA6-B157-7270/part-00000",100) // 100为分片数量
val words = lines.flatMap(_.split(" ")).filter(word => word != " ") // 拆分单词,过滤掉空格
val pairs = words.map(word => (word, 1)) // 对每个单词实例计数为1, 也就是 word => (word, 1)
val wordscount = pairs.reduceByKey(_ + _) // 统计每个单词在文件中出现的总次数, 即key相同的value相加
// val wordscount = pairs.reduceByKey((v1, v2) => v1 + v2) // 等同
//增加广告点击排名,先到过来,再导过去
// val wordscount = pairs.reduceByKey(_ + _).map(pair =>(pair._2,pair._1)).sortByKey(false).map(pair=>(pair._2,pair._1))
//collect 将各个节点数据收集到driver上进行统计
wordscount.collect.foreach(println) // 打印结果,使用collect会将集群中的数据收集到当前运行drive的机器上,需要保证单台机器能放得下所有数据
sc.stop() // 释放资源
}
}
2.程序提交
1.选择导出路径以及配置
2.导出jar包
3.上传至spark服务器并执行,
执行命令:./spark-submit –class com.dt.spark.WordCount unnamed.jar
3.shell脚本
/opt/nsfocus/espc/deps/spark/bin/spark-submit --class com.dt.spark.WordCount unnamed.jar
- 下图中有部分乱码是由于数据源问题,请读者忽略。