当前位置:首页 > 开发 > 系统架构 > 架构 > 正文

转 spark简单实例

发表于: 2015-02-02   作者:blackproof   来源:转载   浏览次数:
摘要: 1、准备文件 2、加载文件 3、显示一行 4、函数运用  (1)map  (2)collecct  (3)filter  (4)flatMap  (5)union  (6) join  (7)lookup  (8)groupByKey  (9)sortBy

1、准备文件

?
1
wget http: //statweb.stanford.edu/~tibs/ElemStatLearn/datasets/spam.data

 

2、加载文件

?
1
scala> val inFile = sc.textFile( "/home/scipio/spam.data" )

 

  输出

?
1
2
3
14 / 06 / 28  12 : 15 : 34  INFO MemoryStore: ensureFreeSpace( 32880 ) called with curMem= 65736 , maxMem= 311387750
14 / 06 / 28  12 : 15 : 34  INFO MemoryStore: Block broadcast_2 stored as values to memory (estimated size  32.1  KB, free  296.9  MB)
inFile: org.apache.spark.rdd.RDD[String] = MappedRDD[ 7 ] at textFile at <console>: 12

 

3、显示一行

?
1
scala> inFile.first()

 

  输出

?
1
2
3
4
5
6
7
8
9
10
14 / 06 / 28  12 : 15 : 39  INFO FileInputFormat: Total input paths to process :  1
14 / 06 / 28  12 : 15 : 39  INFO SparkContext: Starting job: first at <console>: 15
14 / 06 / 28  12 : 15 : 39  INFO DAGScheduler: Got job  0  (first at <console>: 15 ) with  1  output partitions (allowLocal= true )
14 / 06 / 28  12 : 15 : 39  INFO DAGScheduler: Final stage: Stage  0 (first at <console>: 15 )
14 / 06 / 28  12 : 15 : 39  INFO DAGScheduler: Parents of  final  stage: List()
14 / 06 / 28  12 : 15 : 39  INFO DAGScheduler: Missing parents: List()
14 / 06 / 28  12 : 15 : 39  INFO DAGScheduler: Computing the requested partition locally
14 / 06 / 28  12 : 15 : 39  INFO HadoopRDD: Input split: file:/home/scipio/spam.data: 0 + 349170
14 / 06 / 28  12 : 15 : 39  INFO SparkContext: Job finished: first at <console>: 15 , took  0.532360118  s
res2: String =  0  0.64  0.64  0  0.32  0  0  0  0  0  0  0.64  0  0  0  0.32  0  1.29  1.93  0  0.96  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0  0.778  0  0  3.756  61  278  1

 

 

4、函数运用

 (1)map
?
1
2
3
4
5
6
7
8
9
10
11
12
13
scala> val nums = inFile.map(x=>x.split( ' ' ).map(_.toDouble))
nums: org.apache.spark.rdd.RDD[Array[Double]] = MappedRDD[ 8 ] at map at <console>: 14
 
scala> nums.first()
14 / 06 / 28  12 : 19 : 07  INFO SparkContext: Starting job: first at <console>: 17
14 / 06 / 28  12 : 19 : 07  INFO DAGScheduler: Got job  1  (first at <console>: 17 ) with  1  output partitions (allowLocal= true )
14 / 06 / 28  12 : 19 : 07  INFO DAGScheduler: Final stage: Stage  1 (first at <console>: 17 )
14 / 06 / 28  12 : 19 : 07  INFO DAGScheduler: Parents of  final  stage: List()
14 / 06 / 28  12 : 19 : 07  INFO DAGScheduler: Missing parents: List()
14 / 06 / 28  12 : 19 : 07  INFO DAGScheduler: Computing the requested partition locally
14 / 06 / 28  12 : 19 : 07  INFO HadoopRDD: Input split: file:/home/scipio/spam.data: 0 + 349170
14 / 06 / 28  12 : 19 : 07  INFO SparkContext: Job finished: first at <console>: 17 , took  0.011412903  s
res3: Array[Double] = Array( 0.0 0.64 0.64 0.0 0.32 0.0 0.0 0.0 0.0 0.0 0.0 0.64 0.0 0.0 0.0 0.32 0.0 1.29 1.93 0.0 0.96 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.0 0.778 0.0 0.0 3.756 61.0 278.0 1.0 )

 

 (2)collecct
?
1
2
3
4
5
6
7
8
9
scala> val rdd = sc.parallelize(List( 1 , 2 , 3 , 4 , 5 ))
rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[ 9 ] at parallelize at <console>: 12
 
scala> val mapRdd = rdd.map( 2 *_)
mapRdd: org.apache.spark.rdd.RDD[Int] = MappedRDD[ 10 ] at map at <console>: 14
 
scala> mapRdd.collect
14 / 06 / 28  12 : 24 : 45  INFO SparkContext: Job finished: collect at <console>: 17 , took  1.789249751  s
res4: Array[Int] = Array( 2 4 6 8 10 )

 

 (3)filter
?
1
2
3
4
5
6
scala> val filterRdd = sc.parallelize(List( 1 , 2 , 3 , 4 , 5 )).map(_* 2 ).filter(_> 5 )
filterRdd: org.apache.spark.rdd.RDD[Int] = FilteredRDD[ 13 ] at filter at <console>: 12
 
scala> filterRdd.collect
14 / 06 / 28  12 : 27 : 45  INFO SparkContext: Job finished: collect at <console>: 15 , took  0.056086178  s
res5: Array[Int] = Array( 6 8 10 )

 

 (4)flatMap
?
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
scala> val rdd = sc.textFile( "/home/scipio/README.md" )
14 / 06 / 28  12 : 31 : 55  INFO MemoryStore: ensureFreeSpace( 32880 ) called with curMem= 98616 , maxMem= 311387750
14 / 06 / 28  12 : 31 : 55  INFO MemoryStore: Block broadcast_3 stored as values to memory (estimated size  32.1  KB, free  296.8  MB)
rdd: org.apache.spark.rdd.RDD[String] = MappedRDD[ 15 ] at textFile at <console>: 12
 
scala> rdd.count
14 / 06 / 28  12 : 32 : 50  INFO SparkContext: Job finished: count at <console>: 15 , took  0.341167662  s
res6: Long =  127
 
scala> rdd.cache
res7: rdd.type = MappedRDD[ 15 ] at textFile at <console>: 12
 
scala> rdd.count
14 / 06 / 28  12 : 33 : 00  INFO SparkContext: Job finished: count at <console>: 15 , took  0.32015745  s
res8: Long =  127
 
scala> val wordCount = rdd.flatMap(_.split( ' ' )).map(x=>(x, 1 )).reduceByKey(_+_)
wordCount: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[ 20 ] at reduceByKey at <console>: 14
 
scala> wordCount.collect
 
res9: Array[(String, Int)] = Array((means, 1 ), (under, 2 ), ( this , 4 ), (Because, 1 ), (Python, 2 ), (agree, 1 ), (cluster., 1 ), (its, 1 ), (YARN,, 3 ), (have, 2 ), (pre-built, 1 ), (MRv1,, 1 ), (locally., 1 ), (locally, 2 ), (changed, 1 ), (several, 1 ), (only, 1 ), (sc.parallelize( 1 , 1 ), (This, 2 ), (basic, 1 ), (first, 1 ), (requests, 1 ), (documentation, 1 ), (Configuration, 1 ), (MapReduce, 2 ), (without, 1 ), (setting, 1 ), ( "yarn-client" , 1 ), ([params]`., 1 ), (any, 2 ), (application, 1 ), (prefer, 1 ), (SparkPi, 2 ), (<http: //spark.apache.org/>,1), (version,3), (file,1), (documentation,,1), (test,1), (MASTER,1), (entry,1), (example,3), (are,2), (systems.,1), (params,1), (scala>,1), (<artifactId>hadoop-client</artifactId>,1), (refer,1), (configure,1), (Interactive,2), (artifact,1), (can,7), (file's,1), (build,3), (when,2), (2.0.X,,1), (Apac...
 
scala> wordCount.saveAsTextFile( "/home/scipio/wordCountResult.txt" )

 

 (5)union
?
1
2
3
4
5
6
7
8
9
10
11
12
scala> val rdd = sc.parallelize(List(( 'a' , 1 ),( 'a' , 2 )))
rdd: org.apache.spark.rdd.RDD[(Char, Int)] = ParallelCollectionRDD[ 10 ] at parallelize at <console>: 12
 
scala> val rdd2 = sc.parallelize(List(( 'b' , 1 ),( 'b' , 2 )))
rdd2: org.apache.spark.rdd.RDD[(Char, Int)] = ParallelCollectionRDD[ 11 ] at parallelize at <console>: 12
 
scala> rdd union rdd2
res3: org.apache.spark.rdd.RDD[(Char, Int)] = UnionRDD[ 12 ] at union at <console>: 17
 
scala> res3.collect
 
res4: Array[(Char, Int)] = Array((a, 1 ), (a, 2 ), (b, 1 ), (b, 2 ))

 

 (6) join
?
1
2
3
4
5
6
7
8
9
10
11
12
scala> val rdd1 = sc.parallelize(List(( 'a' , 1 ),( 'a' , 2 ),( 'b' , 3 ),( 'b' , 4 )))
rdd1: org.apache.spark.rdd.RDD[(Char, Int)] = ParallelCollectionRDD[ 10 ] at parallelize at <console>: 12
 
scala> val rdd2 = sc.parallelize(List(( 'a' , 5 ),( 'a' , 6 ),( 'b' , 7 ),( 'b' , 8 )))
rdd2: org.apache.spark.rdd.RDD[(Char, Int)] = ParallelCollectionRDD[ 11 ] at parallelize at <console>: 12
 
scala> rdd1 join rdd2
res1: org.apache.spark.rdd.RDD[(Char, (Int, Int))] = FlatMappedValuesRDD[ 14 ] at join at <console>: 17
 
res1.collect
 
res2: Array[(Char, (Int, Int))] = Array((b,( 3 , 7 )), (b,( 3 , 8 )), (b,( 4 , 7 )), (b,( 4 , 8 )), (a,( 1 , 5 )), (a,( 1 , 6 )), (a,( 2 , 5 )), (a,( 2 , 6 )))

 

 (7)lookup
?
1
2
3
val rdd1 = sc.parallelize(List(( 'a' , 1 ),( 'a' , 2 ),( 'b' , 3 ),( 'b' , 4 )))
rdd1.lookup( 'a' )
res3: Seq[Int] = WrappedArray( 1 2 )

 

 (8)groupByKey
?
1
2
3
4
5
val wc = sc.textFile( "/home/scipio/README.md" ).flatMap(_.split( ' ' )).map((_, 1 )).groupByKey
wc.collect
 
14 / 06 / 28  12 : 56 : 14  INFO SparkContext: Job finished: collect at <console>: 15 , took  2.933392093  s
res0: Array[(String, Iterable[Int])] = Array((means,ArrayBuffer( 1 )), (under,ArrayBuffer( 1 1 )), ( this ,ArrayBuffer( 1 1 1 1 )), (Because,ArrayBuffer( 1 )), (Python,ArrayBuffer( 1 1 )), (agree,ArrayBuffer( 1 )), (cluster.,ArrayBuffer( 1 )), (its,ArrayBuffer( 1 )), (YARN,,ArrayBuffer( 1 1 1 )), (have,ArrayBuffer( 1 1 )), (pre-built,ArrayBuffer( 1 )), (MRv1,,ArrayBuffer( 1 )), (locally.,ArrayBuffer( 1 )), (locally,ArrayBuffer( 1 1 )), (changed,ArrayBuffer( 1 )), (sc.parallelize( 1 ,ArrayBuffer( 1 )), (only,ArrayBuffer( 1 )), (several,ArrayBuffer( 1 )), (This,ArrayBuffer( 1 1 )), (basic,ArrayBuffer( 1 )), (first,ArrayBuffer( 1 )), (documentation,ArrayBuffer( 1 )), (Configuration,ArrayBuffer( 1 )), (MapReduce,ArrayBuffer( 1 1 )), (requests,ArrayBuffer( 1 )), (without,ArrayBuffer( 1 )), ( "yarn-client" ,ArrayBuffer( 1 )), ([params]`.,Ar...

 

 (9)sortByKey
?
1
2
3
4
val rdd = sc.textFile( "/home/scipio/README.md" )
val wordcount = rdd.flatMap(_.split( ' ' )).map((_, 1 )).reduceByKey(_+_)
val wcsort = wordcount.map(x => (x._2,x._1)).sortByKey( false ).map(x => (x._2,x._1))
wcsort.saveAsTextFile( "/home/scipio/sort.txt" )

 

 升序的话,sortByKey(true)

http://my.oschina.net/scipio/blog/284957#OSC_h5_11

http://bit1129.iteye.com/blog/2171799

http://bit1129.iteye.com/blog/2171811

转 spark简单实例

  • 0

    开心

    开心

  • 0

    板砖

    板砖

  • 0

    感动

    感动

  • 0

    有用

    有用

  • 0

    疑问

    疑问

  • 0

    难过

    难过

  • 0

    无聊

    无聊

  • 0

    震惊

    震惊

编辑推荐
利用Java编写简单的WebService实例 使用Axis编写WebService比较简单,就我的理解,WebService的实现
本文主要记录Axis开发webservice简单实例的详细过程和步骤: Axis官方网站:http://ws.apache.org/a
转自: http://blog.csdn.net/balance156/article/details/15372831 一、netlink简介 netlink协议是
为了搞清楚Spark Streaming处理数据流,改造了以有的例子来进行测试数据在Spark内部的流向。 packag
SparkStreaming是一个对实时数据流进行高通量、容错处理的流式处理系统,可以对多种数据源(如Kdfka
为了保证高效的数据移动,locality是大数据栈以及分布式应用程序所必须保证的性质,这一点在Spark中
package org.apache.spark.sql.sources import org.apache.spark.SparkContext import java.sql.{Re
本文将建立一个最简单的WCF RIA Services访问数据库的实例,附带有数据库文件,可以在Sql Server中附
ASM是OMF的一个扩展,通过ASM和OMF的结合,oracle在上层将数据文件进行自我管理,在底层将存储进行
《JAVA与模式》之责任链模式 在阎宏博士的《JAVA与模式》一书中开头是这样描述责任链(Chain of Res
版权所有 IT知识库 CopyRight © 2009-2015 IT知识库 IT610.com , All Rights Reserved. 京ICP备09083238号