一 .spark 安装配置
注：按照自己saprk 安装包版本安装
1、下载，解压
$tar zxf spark-1.6.1-bin-2.5.0-cdh5.3.6.tgz
-c /opt/modules/
2、配置
重命名mv spark-env.sh.template spark-env.sh
java_home=/opt/modlues/jdk1.7.0_67
scala_home=/opt/modlues/scala-2.10.4
hadoop_conf_dir=
/opt/modlues/hadoop-2.5.0/etc/hadoop
3、启动hdfs服务
$sbin/hadoop-daemon.sh start namenode
$sbin/hadoop-daemon.sh start datanode
$sbin/start-master.sh
$start-slaves.sh
4、常见问题
命令强制离开
$bin/hadoop dfsadmin -safemode leave
二、实现wordcount
//hdfs path
方式一
val rdd = sc.textfile(“/input/wc.txt”)
//这是在内存中处理 flatmap函数 这是匿名函数
按行读取分割split
val wcrdd=rdd.flatmap(line=>line.split(“\t”)).map(word
=>(word,1)).reducebykey((a,b)=>(a b))
//保存到hdfs
wcrdd.saveastextfile(“wc-spark”)
方式二
val wcrdd=sc.textfile(“/input/wc.txt”).flatmap(line
=>line.split(“\t”)).map(word=>(word,1)).reducebykey((a,b)
=>(a b)).saveastextfile(“wc-spark”)
方式三
val rdd = sc.textfile(“/input/wc.txt”)
//分割这个行
val linerdd=rdd.flatmap(line=>line.split(“\t”))
//元组对出现一个 就是一个元组对
val kvrdd=linerdd.map(word=>(word,1))
val wcrdd=kvrdd.reducebykey((a,b)=>(a b))
wcrdd.saveastextfile(“wc-spark”)
方式四
val rdd = sc.textfile(“/input/wc.txt”)
val wcrdd=rdd.flatmap(_.split(“\t”)).map
((_,1)).reducebykey((_ _))
wcrdd.saveastextfile(“wc-spark”)
spark sc：sparkcontext（上下文）
spark context available as sc.
三、spark常用函数
过滤filter
val rdd = sc.textfile(“/input/wc.txt”).filter
(line=>line.contains(“hadoop”))
val wcrdd=rdd.flatmap(_.split(“\t”)).map
((_,1)).reducebykey((_ _))
循环foreach
wcrdd.foreach(word=>println(word))
//查看函数
wcrdd.first
wcrdd.take(3)
wcrdd.top(3)
wcrdd.collect 查看结果
wcrdd.count
//缓存
wcrdd.cache
wcrdd.count//使缓存生效
四、spark 二次排序
分析：（spark，2）=》（2，spark）》sortbykey(false)=》
（2，spark）=》（spark，2）
代码：
val rdd = sc.textfile(“/input/wc.txt”)
val wcrdd=rdd.flatmap(_.split(“\t”)).map
((_,1)).reducebykey((_ _))
val sortrdd=wcrdd.map(word=>(word._2,word._1)).
sortbykey(false).map(word=>(word._2,word._1))
sortrdd.collect
//top:自带排序
val sortrdd=wcrdd.map(word=>(word._2,word._1)).top(3)
五、spark standalone集群模式的配置
1、sparp-evn
//主节点配置
spark_master_ip=li(主机名)
spark_master_port=7077
spark_master_webui_port=8080
//从节点配置
spark_worker_cores=2
spark_worker_memory=2g
spark_worker_port=7078
spark_worker_webui_port=8081
spark_worker_instances=1
2、slaves
写入hostname名称：自己主机名
3、启动服务
$sbin/start-master.sh
$start-slaves.sh
4、spark-shell常用命令行参数
帮助信息查看：
$bin/spark-shell –help
（1）spark运行在本地模式下：
$bin/spark-shell –master local
或者$bin/spark-shell
//本地
bin/spark-shell \
–class org.apache.spark.examples.sparkpi \
–master local \
–jars lib/spark-examples-1.6.1-hadoop2.5.0-cdh5.3.6.jar \
100
//集群
bin/spark-shell \
–class org.apache.spark.examples.sparkpi \
–master spark://bigdata.beifeng.com:7077 \
–jars lib/spark-examples-1.6.1-hadoop2.5.0-cdh5.3.6.jar \
100
spark-shell和spark-submit
一般执行脚本时使用spark-shell；
一般向集群提交job是使用spark-submit。
spark-shell的本质是spark-submit
spark-shell和spark-submit通知只能使用一个
//本地
bin/spark-submit \
–class org.apache.spark.examples.sparkpi \
–master local \
lib/spark-examples-1.6.1-hadoop2.5.0-cdh5.3.6.jar \
100
//集群
bin/spark-submit \
–class org.apache.spark.examples.sparkpi \
–master spark://bigdata.beifeng.com:7077 \
lib/spark-examples-1.6.1-hadoop2.5.0-cdh5.3.6.jar \
100
//spark开发时模式的使用
一般本地模式适合本地开发测试。集群模式适合生产环境
六、spark日志聚合功能配置
1、spark-env.sh
spark_history_opts=”-dspark.history.fs.logdirectory
=hdfs://192.168.1.1:8020/user/li/spark-events”
注：spark-events目录需要创建
2、spark-default.conf
spark.eventlog.enabled
true
spark.eventlog.dir
hdfs://192.168.1.1:8020/user/li/spark-events
注：spark.eventlog.dir 目录和
-dspark.history.fs.logdirectory保持一致
3、启动history服务
$sbin/start-history-server.sh
