码迷,mamicode.com
首页 > 系统相关 > 详细

Spark Shell简单使用

时间:2017-10-30 14:18:08      阅读:221      评论:0      收藏:0      [点我收藏+]

标签:blank   contains   org   理解   creat   space   方式   uil   pytho   

基础

Spark的shell作为一个强大的交互式数据分析工具,提供了一个简单的方式学习API。它可以使用Scala(在Java虚拟机上运行现有的Java库的一个很好方式)或Python。在Spark目录里使用下面的方式开始运行:

 

[plain] view plain copy
 
  1. ./bin/spark-shell  
在Spark Shell中,有一个专有的SparkContext已经为您创建好了,变量名叫做sc。自己创建的SparkContext将无法工作。可以用--master参数来设置SparkContext要连接的集群,用--jars来设置需要添加到CLASSPATH的jar包,如果有多个jar包,可以使用逗号分隔符连接它们。例如,在一个拥有4核的环境上运行spark-shell,使用:
[plain] view plain copy
 
  1. ./bin/spark-shell --master local[4]  
或在CLASSPATH中添加code.jar,使用:
[plain] view plain copy
 
  1. ./bin/spark-shell --master local[4] --jars code.jar  
可以执行spark-shell --help获取完整的选项列表。 
Spark最主要的抽象是叫Resilient Distributed Dataset(RDD)的弹性分布式集合。RDDs可以使用Hadoop InputFormats(例如HDFS文件)创建,也可以从其他的RDDs转换。让我们在Spark源代码目录里从README.md文本文件中创建一个新的RDD。
[plain] view plain copy
 
  1. scala> val textFile = sc.textFile("file:///home/hadoop/hadoop/spark/README.md")  
  2. 16/07/24 03:30:53 INFO storage.MemoryStore: ensureFreeSpace(217040) called with curMem=321016, maxMem=280248975  
  3. 16/07/24 03:30:53 INFO storage.MemoryStore: Block broadcast_2 stored as values in memory (estimated size 212.0 KB, free 266.8 MB)  
  4. 16/07/24 03:30:53 INFO storage.MemoryStore: ensureFreeSpace(20024) called with curMem=538056, maxMem=280248975  
  5. 16/07/24 03:30:53 INFO storage.MemoryStore: Block broadcast_2_piece0 stored as bytes in memory (estimated size 19.6 KB, free 266.7 MB)  
  6. 16/07/24 03:30:53 INFO storage.BlockManagerInfo: Added broadcast_2_piece0 in memory on localhost:43303 (size: 19.6 KB, free: 267.2 MB)  
  7. 16/07/24 03:30:53 INFO spark.SparkContext: Created broadcast 2 from textFile at <console>:21  
  8. textFile: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[5] at textFile at <console>:21  
注意:1. 其中2~7行是日志信息,暂且不必关注,主要看最后一行。之后的运行日志信息将不再贴出。用户也可以进入到spark目录/conf文件夹下,此时有一个log4j.properties.template文件,我们执行如下命令将其拷贝一份为log4j.properties,并对log4j.properties文件进行修改。

 

 

[plain] view plain copy
 
  1. cp log4j.properties.template log4j.properties  
  2. vim log4j.properties  
如下图所示,将INFO改为WARN,这样就不输出蓝色部分的日志信息:

 

技术分享

2. 另外,file:///home/hadoop/hadoop/spark/README.md,首部的file代表本地目录,注意file:后有三个斜杠(/);中间红色部分是我的spark安装目录,读者可根据自己的情况进行替换。

 

RDD的actions从RDD中返回值,transformations可以转换成一个新RDD并返回它的引用。下面展示几个action:

 

[plain] view plain copy
 
  1. scala> textFile.count()  
  2. res0: Long = 98  
  3.   
  4. scala> textFile.first()  
  5. res1: String = # Apache Spark  
其中,count代表RDD中的总数据条数;first代表RDD中的第一行数据。

 

下面使用一个transformation,我们将使用filter函数对textFile这个RDD进行过滤,取出包含字符串"Spark"的行,并返回一个新的RDD:

 

[plain] view plain copy
 
  1. scala> val linesWithSpark = textFile.filter(line => line.contains("Spark"))  
  2. linesWithSpark: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at filter at <console>:23  
当然也可以把actions和transformations连接在一起使用:

 

 

[plain] view plain copy
 
  1. scala> textFile.filter(line => line.contains("Spark")).count()  
  2. res2: Long = 19  
上面这条语句表示有多少行包括字符串"Spark"。

 

 

更多RDD操作

RDD actions和transformations能被用在更多的复杂计算中。比如想要找到一行中最多的单词数量:

 

[plain] view plain copy
 
  1. scala> textFile.map(line => line.split(" ").size).reduce((a, b) => if (a > b) a else b)  
  2. res3: Int = 14  
首先将行映射成一个整型数值产生一个新的RDD。在这个新的RDD上调用reduce找到行中最大的单词数个数。map和reduce的参数是Scala的函数串(闭包),并且可以使用任何语言特性或者Scala/Java类库。例如,我们可以很方便地调用其他的函数声明。我们使用Math.max()函数让代码更容易理解:

 

 

[plain] view plain copy
 
  1. scala> import java.lang.Math  
  2. import java.lang.Math  
  3.   
  4. scala> textFile.map(line => line.split(" ").size).reduce((a, b) => Math.max(a, b))  
  5. res4: Int = 14  
大家都知道,Hadoop流行的一个通用数据流模式是MapReduce。Spark能够很容易地实现MapReduce:

 

 

[plain] view plain copy
 
  1. scala> val wordCounts = textFile.flatMap(line => line.split(" ")).map(word => (word, 1)).reduceByKey((a, b) => a + b)  
  2. wordCounts: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[8] at reduceByKey at <console>:24  
这里,我们结合了flatMap、map和reduceByKey来计算文件里每个单词出现的数量,它的结果是包含一组(String, Int)键值对的RDD。我们可以使用collect操作收集单词的数量:

 

 

[plain] view plain copy
 
  1. scala> wordCounts.collect()  
  2. res5: Array[(String, Int)] = Array((package,1), (For,2), (Programs,1), (processing.,1), (Because,1), (The,1), (cluster.,1), (its,1), ([run,1), (APIs,1), (have,1), (Try,1), (computation,1), (through,1), (several,1), (This,2), ("yarn-cluster",1), (graph,1), (Hive,2), (storage,1), (["Specifying,1), (To,2), (page](http://spark.apache.org/documentation.html),1), (Once,1), (application,1), (prefer,1), (SparkPi,2), (engine,1), (version,1), (file,1), (documentation,,1), (processing,,2), (the,21), (are,1), (systems.,1), (params,1), (not,1), (different,1), (refer,2), (Interactive,2), (given.,1), (if,4), (build,3), (when,1), (be,2), (Tests,1), (Apache,1), (all,1), (./bin/run-example,2), (programs,,1), (including,3), (Spark.,1), (package.,1), (1000).count(),1), (Versions,1), (HDFS,1), (Data.,1), (>...  

 

 

缓存

Spark支持把数据集缓存到内存之中,当要重复访问时,这是非常有用的。举一个简单的例子:

[plain] view plain copy
 
  1. scala> linesWithSpark.cache()  
  2. res6: linesWithSpark.type = MapPartitionsRDD[2] at filter at <console>:23  
  3.   
  4. scala> linesWithSpark.count()  
  5. res7: Long = 19  
  6.   
  7. scala> linesWithSpark.count()  
  8. res8: Long = 19  
  9.   
  10. scala> linesWithSpark.count()  
  11. res9: Long = 19  
首先缓存linesWithSpark数据集,然后重复访问count函数返回的值。当然,我们并不能察觉明显的查询速度变化,但是当在大型的数据集中使用缓存,将会非常显著的提升相应的迭代操作速度。

Spark Shell简单使用

标签:blank   contains   org   理解   creat   space   方式   uil   pytho   

原文地址:http://www.cnblogs.com/csguo/p/7753565.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!