spark笔记

chengjianxiaoxue

浏览: 1310627 次
性别:
来自: 北京

最近访客更多访客>>

liu_shui8

happy2012

nddht

yhtppp

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

spark

1 版本信息:

spark1.4.1 官网选择了spark1.4.1版本后，会看到默认的 pre-built for hadoop2.6

scala 2.10.4

jdk 1.7

2 几个图：

spark是计算框架，没有存储框架，存储这块需要用到hdfs 计算这块能搞定 实时查询+流式计算+批量计算
 
分组和分区的区别：
分到一个班等于分到一个组，我们要共同学习一个知识点所以分到同一个组中。分组中同组内的key是相同的 是解决key的分配问题
分区： 你要打到哪个reduce上去 分区里一个区中可能有不同的key 是解决数据的分配存储问题

 
为啥spark在机器计算时优势 ：
数据共享对比: 在使用mr计算时，除了要有shuffle这种多次落地hdfs文件数据开销外，每份文件都有3个备份也是一种读写开销，
并且在机器学习中，很多场景是这次计算结果是下次迭代的开始，很频繁，用mr在机器学习上也会慢。
 

1 spark 在window下启动和执行：

window环境下已经按照了 jdk，可以安装scala可以不安装下载解压的spark就自带scala
spark下载后解压在window的目录不要有 中文和单词之间不要有空格
在D:\spark-1.4.1-bin-hadoop2.6\bin下 执行: spark-shell.cmd
启动后cmd界面如下： 启动后自动进入scala的命令行
展示 sparkUI 端口

启动成功后样子：

 

4 按照上面提示，输入链接： http://localhost:4040/

5 准备单词技术的文本：

D:\spark-wodcount\1.txt 其内容为

6 在scala 命令行内输入命令：

sc.textFile("d://spark-wodcount//*").flatMap(_.split("\t")).map((_,1)).reduceByKey(_ + _).collect

单词统计结果如下：

idea中导入spark包执行单词计数：

spark执行单词计数命令解释:

scala>sc.textFile("d://test/1.txt").collect ---> 加载数据 spark是懒执行机制，其中加载的数据d://test/1.txt会变成一个rdd,调用collect进行计算

上述命令展示结果：

res2: Array[String] = Array(a b,c) spark像hdfs读取数据，每次进来都是一行数据，每一行作为一个元素，1.txt内容 a b 一行 c为第二行，而这两行构建了scala的两个元素

scala>sc.textFile("d://test/*").collect 加载多个文件

单词计数实例

sc.textFile("d://test//*").flatMap(_.split("\t")).map((_,1)).reduceByKey(_ + _).collect

单词计数完整版

sc.textFile("d://test//*").flatMap((line:String) => line.split("\t")).map((x:String)=>(x,1)).reduceByKey((x:Int,y:Int)=>(x+y)).collect

使用内存大小：

impala > spark

linux上启动 spark :

在linux上按照 spark，官网上下载好了 spark后，解压，将 conf下的/ spark-env.sh.template mv 成 spark-env.sh

spark-shell.sh 在linux上启动 spark

spark在Linux上运行分为如下几种方式：

1 local方式：

#./spark-shell ----> local ----使用一个工作线程来运行计算任务,不会重新计算失败的任务

#MASTER=local[2] ./spark-shell ----> local ----使用指定的线程运行计算任务，不会重新计算失败的任务。

#MASTER=local-cluster[2,2,1024] ./spark-shell ----> 在一台机器上模拟伪分布模式， [节点个数,每个节点核数,每个节点内存MB] webui端口: 44394

standalone架构

Master:负责任务分发；负责分配资源；

Worker:运行应用代码的节点,接收Master请求，启动任务；向Master汇报信息，资源信息和健康状况。

Executor:worker node的一个进程,接收worker的任务执行请求，执行任务；向worker汇报任务执行情况。

Driver:提交任务；监控任务；

读取本地linux数据

sc.textFile("/root/wctext.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_ + _).collect（需要每台机器都有该数据）

读取远程hdfs数据并保存到远程hdfs

sc.textFile("hdfs://192.168.1.88:9000/wctext").flatMap(_.split(" ")).map((_,1)).reduceByKey(_ + _).saveAsTextFile("hdfs://192.168.1.88:9000/0828out")

sc.textFile("hdfs://192.168.1.88:9000/wctext").flatMap(_.split(" ")).map((_,1)).reduceByKey(_ + _).repartition(1).saveAsTextFile("hdfs://192.168.1.88:9000/0828out")

// 设置输出只有一个输出分区

sc.textFile("hdfs://192.168.1.88:9000/wctext",1).flatMap(_.split(" ")).map((_,1)).reduceByKey(_ + _).saveAsTextFile("hdfs://192.168.1.88:9000/0828out")

sc.textFile("hdfs://192.168.1.88:9000/wctext").flatMap(_.split(" ")).map((_,1)).reduceByKey(_ + _).repartition(1).saveAsTextFile("hdfs://192.168.1.88:9000/0828out")

当然可以使用这种方式：表示输出只有一个分区，输出到hdfs的 wc2.txt 文件夹下

scala> sc.textFile("/test/wc.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).repartition(1).saveAsTextFile("/test/wc2.txt")

打包后上传到服务器后，执行命令：

./bin/spark-submit --class xxx --master spark://ip:7077 xxx.jar 参数