生态
特点
体系架构
安装部署
安装jdk,配置主机名,配置免密码登录
伪分布(Standalone):一台机器上模拟分布式环境(Master+Worker)
核心配置文件:conf/spark-env.sh
启动:sbin/start-all.sh
Web Console:http://192.168.174.111:8080/
全分布:先在主节点上安装,再把装好的目录复制到从节点上
HA
基于文件目录
基于zookeeper
工具
spark-submit:用于提交Spark任务(jar包)
spark-shell:相当于REPL,命令行工具
本地模式
集群模式
IDE开发WordCount
Scala版本
1 package day0605
2
3 import org.apache.spark.SparkConf
4 import org.apache.spark.SparkContext
5
6 object MyWordCount {
7 def main(args:Array[String]):Unit = {
8 //创建一个任务的配置信息
9 //设置Master=local,表示运行在本地模式上
10 //集群模式不需设置Master
11 val conf = new SparkConf().setAppName("MyWordCount").setMaster("local")
12
13 //创建一个SparkContext对象
14 val sc = new SparkContext(conf)
15
16 //执行WordCount
17 val result = sc.textFile("hdfs://192.168.174.111:9000/input/data.txt")
18 .flatMap(_.split(" ")).map((_,1))
19 .reduceByKey(_+_).collect
20
21 //打印结果
22 result.foreach(println)
23
24 //停止SparkContext
25 sc.stop()
26 }
27 }
* 集群模式
* bin/spark-submit --master spark://bigdata111:7077 --class day0605.MyWordCount /root/temp/demo1.jar hdfs://bigdata111:9000/input/data.txt hdfs://bigdata111:9000/output/0605/demo1
1 package day0605
2
3 import org.apache.spark.SparkConf
4 import org.apache.spark.SparkContext
5
6 //通过spark-submit提交
7
8 object MyWordCount {
9 def main(args:Array[String]):Unit = {
10 //创建一个任务的配置信息
11 //设置Master=local,表示运行在本地模式上
12 //集群模式不需设置Master
13 val conf = new SparkConf().setAppName("MyWordCount")
14
15 //创建一个SparkContext对象
16 val sc = new SparkContext(conf)
17
18 //执行WordCount
19 val result = sc.textFile(args(0))
20 .flatMap(_.split(" "))
21 .map((_,1))
22 .reduceByKey(_+_)
23
24 //输出到hdfs
25 result.saveAsTextFile(args(1))
26
27 //停止SparkContext
28 sc.stop()
29 }
30 }
1 package demo;
2
3 import java.util.Arrays;
4 import java.util.Iterator;
5 import java.util.List;
6
7 import org.apache.spark.SparkConf;
8 import org.apache.spark.api.java.JavaPairRDD;
9 import org.apache.spark.api.java.JavaRDD;
10 import org.apache.spark.api.java.JavaSparkContext;
11 import org.apache.spark.api.java.function.FlatMapFunction;
12 import org.apache.spark.api.java.function.Function2;
13 import org.apache.spark.api.java.function.PairFunction;
14
15 import scala.Tuple2;
16
17 /*
18 * 使用spark submit提交
19 * bin/spark-submit --master spark://bigdata111:7077 --class demo.JavaWordCount /root/temp/demo2.jar hdfs://bigdata111:9000/input/data.txt
20 */
21
22 public class JavaWordCount {
23
24 public static void main(String[] args) {
25 //运行在本地模式,可以设置断点
26 SparkConf conf = new SparkConf().setAppName("JavaWordCount").setMaster("local");
27
28 //运行在集群模式
29 //SparkConf conf = new SparkConf().setAppName("JavaWordCount");
30
31 //创建一个SparkContext对象: JavaSparkContext对象
32 JavaSparkContext sc = new JavaSparkContext(conf);
33
34 //读入HDFS的数据
35 JavaRDD
36
37 /*
38 * 分词
39 * FlatMapFunction:接口,用于处理分词的操作
40 * 泛型:String 读入的每一句话
41 * U: 返回值 ---> String 单词
42 */
43 JavaRDD
44
45 @Override
46 public Iterator
47 //数据: I love Beijing
48 //分词
49 return Arrays.asList(input.split(" ")).iterator();
50 }
51 });
52
53 /*
54 * 每个单词记一次数 (k2 v2)
55 * Beijing ---> (Beijing,1)
56 * 参数:
57 * String:单词
58 * k2 v2不解释
59 */
60 JavaPairRDD
61
62 @Override
63 public Tuple2
64 return new Tuple2
65 }
66
67 });
68
69 //执行Reduce的操作
70 JavaPairRDD
71
72 @Override
73 public Integer call(Integer a, Integer b) throws Exception {
74 //累加
75 return a+b;
76 }
77 });
78
79 //执行计算(Action),把结果打印在屏幕上
80 List
81
82 for(Tuple2
83 System.out.println(tuple._1+"\t"+tuple._2);
84 }
85
86 //停止JavaSparkContext对象
87 sc.stop();
88 }
89 }
参考
spark.apache.org
spark任务提交两种方式
手机扫一扫
移动阅读更方便
你可能感兴趣的文章