Spark

---spark

spark 产生背景

MapReduce的局限性

1）代码繁琐
2）只能够支持map与reduce的方法
3）执行效率低下
4）不适合迭代多次（机器学习）、交互式（数据挖掘）、流式的处理（MapReduice数据源为静态）

框架多样化

1）批处理(离线):MapReduce Hive Pig
2）流式处理： Storm、JStorm
3）交互式计算：Impala

=======> Spark

Spark 生态圈
BDAS ： Berkeley Data Analytics Stack

spark搭建文档:http://spark.apache.org/docs/latest/building-spark.html
Spark搭建环境
首先进入官网下载spark-2.1.0.tgz 并且解压

1）前提Maven大于3.3.9 jdk大于7 目前最新2.3需要8+
2）export MAVEN_OPTS="-Xmx2g -XX:ReservedCodeCacheSize=512m" 您需要通过设置来配置Maven以使用比平时更多的内存MAVEN_OPTS：
如果是cdh版本
进入 pom.xml在原来的<repositories>内增加（要放到原有仓库的后面）
<repository>
<id>central</id>
<url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
</repository>
进入spark根目录
Maven编译命令 指定依赖与hadoop版本（如果hadoop不设置 默认2.2.0）
./build/mvn -Pyarn -Phadoop-2.6 -Dhadoop.version=2.6.0-cdh5.7.0 -Phive -Phive-thriftserver -DskipTests clean package
（耗时较长 可能达到小时级别）

Maven编译可运行的 #推荐使用
先指定scala版本
./dev/change-scala-version.sh 2.x.x

./dev/make-distribution.sh
--name 2.6.0-cdh5.7.0
--tgz
-Pyarn -Phadoop-2.6
-Phive -Phive-thriftserver
-Dhadoop.version=2.6.0-cdh5.7.0
-U
-X

具体可以看make-distribution.sh脚本 这种编译模式可以不需要配置内存 脚本已经存在
如果报内存不够 请自行加内存。
编译完成后:
spark-VERSION-bin-NAME.tgz
我的名称 spark-2.1.0-bin-2.6.0-cdh5.7.0.tgz 然后解压就可以了

启动spark local模式
首先进入根目录的bin文件夹 可以不带jars 我个人需要这个jar包
./spark-shell --master local[2] --jars ~/software/mysql-connector-java-5.1.27-bin.jar

现在可以打开spark的web界面
http://ip:4040/jobs/

spark Standalone模式的架构和Hadoop 与HDFS/YARN类似
1 master + n worker
进入spark根目录
cd conf
vi spark-env.sh
SPARK_MASTER_HOST=hadoop001
SPARK_WORKER_CORES=2
SPARK_WORKER_MEMORY=2g
SPARK_WORKER_INSTANCES=1
然后进入sbin
./start-all.sh
查看 logs/spark-hadoop-org.apache.spark.deploy.master.Master-1-hadoop001.out
logs中包含了很多启动日志 可以看到端口的占用

可以在http://ip:8080/ 中查看master状态

多台机器的情况
可以配置
vi conf/slaves
每行加入一台hostname

这样当我们启动start-all.sh时 会在slaves列表中的机器 去启动woker进程

启动spark-shell
./spark-shell --master spark://hadoop001:7077 //master 为我们的master host给出的webUI上的Master URL

spark-shell 中实现wordcount
val file = spark.sparkContext.textFile("file:///home/hadoop/data/wc.txt")
val wordCounts = file.flatMap(line => line.split(",")).map((word =>(word,1))).reduceByKey(+)
wordCounts.collect

wc.txt 中的内容
hello,world,hello
hello,world
welcome