Spark第一个程序开发 wordcount
2016-07-24 12:17
393 查看
这里介绍了程序运行在本地模式和Standalone模式两种方式
如果运行在集群中,需要将程序打包成jar包,使用spark提供的spark-submit提交到集群
–class: 应用入口类(例如:org.apache.spark.examples.SparkPi
))
–master: 集群的master URL (如:spark://23.195.26.187:7077)
更多提交时设置的内容可以参考Spark官方文档http://spark.apache.org/docs/latest/submitting-applications.html
Scala
package com.spark.app import org.apache.spark.{SparkContext, SparkConf} /** * Created by Administrator on 2016/7/24 0024. */ object WordCount { def main(args: Array[String]) { /** * 第1步;创建Spark的配置对象SparkConf,设置Spark程序运行时的配置信息 * 例如 setAppName用来设置应用程序的名称,在程序运行的监控界面可以看到该名称, * setMaster设置程序运行在本地还是运行在集群中,运行在本地可是使用local参数,也可以使用local[K]/local[*], * 可以去spark官网查看它们不同的意义。 如果要运行在集群中,以Standalone模式运行的话,需要使用spark://HOST:PORT * 的形式指定master的IP和端口号,默认是7077 */ val conf = new SparkConf().setAppName("WordCount").setMaster("local") // val conf = new SparkConf().setAppName("WordCount").setMaster("spark://master:7077") // 运行在集群中 /** * 第2步:创建SparkContext 对象 * SparkContext是Spark程序所有功能的唯一入口 * SparkContext核心作用: 初始化Spark应用程序运行所需要的核心组件,包括DAGScheduler、TaskScheduler、SchedulerBackend * 同时还会负责Spark程序往Master注册程序 * * 通过传入SparkConf实例来定制Spark运行的具体参数和配置信息 */ val sc = new SparkContext(conf) /** * 第3步: 根据具体的数据来源(HDFS、 HBase、Local FS、DB、 S3等)通过SparkContext来创建RDD * RDD 的创建基本有三种方式: 根据外部的数据来源(例如HDFS)、根据Scala集合使用SparkContext的parallelize方法、 * 由其他的RDD操作产生 * 数据会被RDD划分成为一系列的Partitions,分配到每个Partition的数据属于一个Task的处理范畴 */ val lines = sc.textFile("D:/resources/README.md") // 读取本地文件 // val lines = sc.textFile("/library/wordcount/input") // 读取HDFS文件,并切分成不同的Partition // val lines = sc.textFile("hdfs://master:9000/libarary/wordcount/input") // 或者明确指明是从HDFS上获取数据 /** * 第4步: 对初始的RDD进行Transformation级别的处理,例如 map、filter等高阶函数来进行具体的数据计算 */ val words = lines.flatMap(_.split(" ")).filter(word => word != " ") // 拆分单词,并过滤掉空格,当然还可以继续进行过滤,如去掉标点符号 val pairs = words.map(word => (word, 1)) // 在单词拆分的基础上对每个单词实例计数为1, 也就是 word => (word, 1) val wordscount = pairs.reduceByKey(_ + _) // 在每个单词实例计数为1的基础之上统计每个单词在文件中出现的总次数, 即key相同的value相加 // val wordscount = pairs.reduceByKey((v1, v2) => v1 + v2) // 等同于 wordscount.collect.foreach(println) // 打印结果,使用collect会将集群中的数据收集到当前运行drive的机器上,需要保证单台机器能放得下所有数据 sc.stop() // 释放资源 } }
如果运行在集群中,需要将程序打包成jar包,使用spark提供的spark-submit提交到集群
spark-submit --class com.spark.app.WordCount --master spark://master:7077 /root/Documents/SparkApps/wordcount.jar
–class: 应用入口类(例如:org.apache.spark.examples.SparkPi
))
–master: 集群的master URL (如:spark://23.195.26.187:7077)
更多提交时设置的内容可以参考Spark官方文档http://spark.apache.org/docs/latest/submitting-applications.html
通过sortByKey方法对单词出现的次序进行排序:
package com.spark.app import org.apache.spark.{SparkContext, SparkConf} /** * Created by Administrator on 2016/7/24 0024. */ object WordCountSorted { def main(args: Array[String]) { def conf = new SparkConf().setAppName("WordCountSorted").setMaster("local") def sc = new SparkContext(conf) val lines = sc.textFile("D:/resources/README.md") val words = lines.flatMap(_.split(" ")).filter(word => word != " ") val pairs = words.map(word => (word, 1)) /** * 在这里通过reduceByKey方法之后可以获得每个单词出现的次数 * 第一个map将单词和出现的次数交换,将出现的次数作为key,使用sortByKey进行排序(false为降序) * 第二个map将出现的次数和单词交换,这样还是恢复到以单词作为key */ val wordcount = pairs.reduceByKey(_ + _).map(pair => (pair._2, pair._1)).sortByKey(false).map(pair => (pair._2, pair._1)) wordcount.collect.foreach(println) sc.stop() } }
Java 版本
import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaPairRDD; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.*; import scala.Tuple2; import java.util.Arrays; /** * Created by Administrator on 2016/7/24 0024. */ public class WordCount { public static void main(String[] args) { SparkConf conf = new SparkConf().setAppName("WordCount").setMaster("local"); JavaSparkContext sc = new JavaSparkContext(conf); JavaRDD<String> lines = sc.textFile("D:/resources/README.md"); JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() { @Override public Iterable<String> call(String line) throws Exception { return Arrays.asList(line.split(" ")); } }); JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() { @Override public Tuple2<String, Integer> call(String word) throws Exception { return new Tuple2<String, Integer>(word, 1); } }); JavaPairRDD<String, Integer> wordcount = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() { @Override public Integer call(Integer v1, Integer v2) throws Exception { return v1 + v2; } }); for (Tuple2<String, Integer> pair : wordcount.collect()) { System.out.println(pair._1 + ": " + pair._2); } sc.close(); } }
相关文章推荐
- Spark RDD API详解(一) Map和Reduce
- 使用spark和spark mllib进行股票预测
- Spark随谈——开发指南(译)
- Spark,一种快速数据分析替代方案
- eclipse 开发 spark Streaming wordCount
- Understanding Spark Caching
- ClassNotFoundException:scala.PreDef$
- Windows 下Spark 快速搭建Spark源码阅读环境
- Spark中将对象序列化存储到hdfs
- 使用java代码提交Spark的hive sql任务,run as java application
- Spark机器学习(一) -- Machine Learning Library (MLlib)
- Spark机器学习(二) 局部向量 Local-- Data Types - MLlib
- Spark机器学习(三) Labeled point-- Data Types
- Spark初探
- Spark Streaming初探
- Spark本地开发环境搭建
- 搭建hadoop/spark集群环境
- Spark HA部署方案
- Spark HA原理架构图
- spark内存概述