Spark First Sample Demo

我会带着你远行 2022-08-08 00:50 177阅读 0赞

## 1. 运行环境 ##

*  OS:

\[root@karei spark-1.3.1-bin-hadoop2.6\]\# uname -a  
Linux karei 2.6.18-371.4.1.el5 \#1 SMP Wed Jan 8 18:42:07 EST 2014 x86\_64 x86\_64 x86\_64 GNU/Linux  
\[root@karei spark-1.3.1-bin-hadoop2.6\]\#

*  CPU:

\[root@karei spark-1.3.1-bin-hadoop2.6\]\# cat /proc/cpuinfo | grep name | cut -d':' -f2 | uniq -c  
      4  Intel(R) Xeon(R) CPU           X5270  @ 3.50GHz  
\[root@karei spark-1.3.1-bin-hadoop2.6\]\#

\[root@karei spark-1.3.1-bin-hadoop2.6\]\#   
\[root@karei spark-1.3.1-bin-hadoop2.6\]\# cat /proc/cpuinfo | grep "physical id" | uniq -c  
      1 physical id     : 0  
      1 physical id     : 3  
      1 physical id     : 0  
      1 physical id     : 3  
\[root@karei spark-1.3.1-bin-hadoop2.6\]\#

*  Memory:

\[root@karei spark-1.3.1-bin-hadoop2.6\]\#   
\[root@karei spark-1.3.1-bin-hadoop2.6\]\# grep MemTotal /proc/meminfo  
MemTotal:     32961052 kB  
\[root@karei spark-1.3.1-bin-hadoop2.6\]\#   
\[root@karei spark-1.3.1-bin-hadoop2.6\]\#

## 2. 下载Spark ##

我用的是spark-1.3.1-bin-hadoop2.6.tgz

[http://spark.apache.org/downloads.html][http_spark.apache.org_downloads.html]

## 3. Linux上解压Spark压缩包 ##

\[root@karei opt\]\# ll /opt/spark-1.3.1-bin-hadoop2.6/  
total 460  
\-rwxrw-r-- 1 1000 1000 278851 Apr 11 08:32 CHANGES.txt  
\-rwxrw-r-- 1 1000 1000  46083 Apr 11 08:32 LICENSE  
\-rwxrw-r-- 1 1000 1000  22559 Apr 11 08:32 NOTICE  
\-rwxrw-r-- 1 1000 1000   3629 Apr 11 08:32 README.md  
\-rwxrw-r-- 1 1000 1000    157 Apr 11 08:32 RELEASE  
drwxrwxr-x 2 1000 1000   4096 Apr 11 08:32 bin  
drwxrwxr-x 2 1000 1000   4096 Apr 11 08:32 conf  
drwxrwxr-x 3 1000 1000   4096 Apr 11 08:32 data  
drwxrwxr-x 3 1000 1000   4096 Apr 11 08:32 ec2  
drwxrwxr-x 3 1000 1000   4096 Apr 11 08:32 examples  
drwxrwxr-x 2 1000 1000   4096 Apr 11 08:32 lib  
drwxr-xr-x 2 root root   4096 Apr 27 10:49 mylib  
drwxrwxr-x 7 1000 1000   4096 Apr 11 08:32 python  
drwxrwxr-x 2 1000 1000   4096 Apr 11 08:32 sbin  
drwxr-xr-x 4 root root   4096 Apr 27 11:10 tmp  
\[root@karei opt\]\#

## 4. Linux上配置Spark环境变量 ##

\[root@karei opt\]\# export PATH=/opt/spark-1.3.1-bin-hadoop2.6/bin:$PATH

## 5. Spark Sample开发 ##

Maven(3.2.3) + Eclipse(Luna 4.4.1) + Spark(1.3.1) + JDK(1.8.45)

*  pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    	<modelVersion>4.0.0</modelVersion>
    
    	<groupId>shuai.spark</groupId>
    	<artifactId>spark-demo</artifactId>
    	<version>1.0</version>
    	<packaging>jar</packaging>
    
    	<name>spark-demo</name>
    	<url>http://maven.apache.org</url>
    
    	<properties>
    		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    		<spark.version>1.3.1</spark.version>
    		<junit.version>4.11</junit.version>
    		<maven.compiler.source>1.8</maven.compiler.source>
    		<maven.compiler.target>1.8</maven.compiler.target>
    	</properties>
    
    	<dependencies>
    		<dependency>
    			<groupId>org.apache.commons</groupId>
    			<artifactId>commons-lang3</artifactId>
    			<version>3.1</version>
    		</dependency>
    		<dependency>
    			<groupId>commons-io</groupId>
    			<artifactId>commons-io</artifactId>
    			<version>2.4</version>
    		</dependency>
    
    		<dependency>
    			<groupId>org.apache.spark</groupId>
    			<artifactId>spark-core_2.10</artifactId>
    			<version>${spark.version}</version>
    		</dependency>
    
    		<dependency>
    			<groupId>junit</groupId>
    			<artifactId>junit</artifactId>
    			<scope>test</scope>
    			<version>${junit.version}</version>
    		</dependency>
    		<dependency>
    			<groupId>org.hamcrest</groupId>
    			<artifactId>hamcrest-all</artifactId>
    			<scope>test</scope>
    			<version>1.3</version>
    		</dependency>
    	</dependencies>
    
    	<build>
    		<plugins>
    			<plugin>
    				<groupId>org.apache.maven.plugins</groupId>
    				<artifactId>maven-compiler-plugin</artifactId>
    				<version>3.3</version>
    			</plugin>
    
    			
    		</plugins>
    	</build>
    </project>

*  SparkDemo.java

package shuai.spark;
    
    import java.io.File;
    import java.util.Arrays;
    
    import org.apache.commons.io.FileUtils;
    import org.apache.spark.SparkConf;
    import org.apache.spark.api.java.JavaPairRDD;
    import org.apache.spark.api.java.JavaRDD;
    import org.apache.spark.api.java.JavaSparkContext;
    import org.apache.spark.api.java.function.FlatMapFunction;
    import org.apache.spark.api.java.function.Function;
    import org.apache.spark.api.java.function.Function2;
    import org.apache.spark.api.java.function.PairFunction;
    
    import scala.Tuple2;
    
    public class SparkDemo {
    	private static JavaSparkContext javaSparkContext = null;
    
    	public static void init(String appName, String master) {
    		SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);
    		javaSparkContext = new JavaSparkContext(conf);
    	}
    
    	@SuppressWarnings("serial")
    	private static void wordCount(String filePath, String fileDir) {
    		FileUtils.deleteQuietly(new File(fileDir));
    
    		JavaRDD<String> file = javaSparkContext.textFile(filePath);
    
    		JavaRDD<String> words = file.flatMap(new FlatMapFunction<String, String>() {
    			@Override
    			public Iterable<String> call(String s) {
    				return Arrays.asList(s.split(" "));
    			}
    		});
    
    		JavaPairRDD<String, Integer> pairs = words.mapToPair(new PairFunction<String, String, Integer>() {
    			@Override
    			public Tuple2<String, Integer> call(String s) {
    				return new Tuple2<String, Integer>(s, 1);
    			}
    		});
    
    		JavaPairRDD<String, Integer> counts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() {
    			@Override
    			public Integer call(Integer a, Integer b) {
    				return a + b;
    			}
    		});
    
    		counts.saveAsTextFile(fileDir);
    	}
    
    	@SuppressWarnings("serial")
    	private static void errorCount(String filePath) {
    		JavaRDD<String> file = javaSparkContext.textFile(filePath);
    
    		JavaRDD<String> errors = file.filter(new Function<String, Boolean>() {
    			@Override
    			public Boolean call(String s) {
    				return s.contains("ERROR");
    			}
    		});
    
    		errors.count();
    
    		errors.filter(new Function<String, Boolean>() {
    			@Override
    			public Boolean call(String s) {
    				return s.contains("ORACLE");
    			}
    		}).count();
    
    		errors.filter(new Function<String, Boolean>() {
    			@Override
    			public Boolean call(String s) {
    				return s.contains("ORACLE");
    			}
    		}).collect();
    	}
    
    	public static void main(String[] args) {
    		if (args.length < 2) {
    			System.err.println("Please provide the input file full path and output dir as argument");
    			System.exit(0);
    		}
    
    		SparkDemo.init("shuai.spark", "local");
    
    		SparkDemo.wordCount(args[0], args[1]);
    
    		SparkDemo.errorCount(args[0]);
    	}
    }

## 6. Linux上部署Spark ##

创建相应的目录和文件，并将spark-demo-1.0.jar部署到环境上

\[root@karei mylib\]\#  
\[root@karei mylib\]\# ll /usr/local/spark/mylib/spark-demo-1.0.jar  
\-rwxr--r-- 1 spark bigdata 9027 Apr 27 11:05 /usr/local/spark/mylib/spark-demo-1.0.jar  
\[root@karei mylib\]\#  
\[root@karei mylib\]\# ll /home/hadoop/input/test.log  
\-rw-r--r-- 1 root root 119 Apr 30 08:14 /home/hadoop/input/test.log  
\[root@karei mylib\]\#

[root@karei spark]# cat /home/hadoop/input/test.log
    INFO : Spark testing start
    WARN : Hadoop HDFS Configuration
    ERROR : Database ORACLE Operation
    INFO : Spark testing end
    [root@karei spark]#

## 7. Sample运行 ##

\[root@karei spark\]\#  
\[root@karei spark\]\# ./bin/spark-submit --class shuai.spark.SparkDemo --master local\[2\] mylib/spark-demo-1.0.jar /home/hadoop/input/test.log /home/hadoop/output  
\[root@karei spark\]\#

[root@karei spark]#
    [root@karei spark]# ./bin/spark-submit --class shuai.spark.SparkDemo --master local[2] mylib/spark-demo-1.0.jar /home/hadoop/input/test.log /home/hadoop/output
    15/04/30 08:24:39 INFO spark.SparkContext: Running Spark version 1.3.1
    15/04/30 08:24:39 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
    15/04/30 08:24:39 INFO spark.SecurityManager: Changing view acls to: root
    15/04/30 08:24:39 INFO spark.SecurityManager: Changing modify acls to: root
    15/04/30 08:24:39 INFO spark.SecurityManager: SecurityManager: authentication disabled; ui acls disabled; users with view permissions: Set(root); users with modify permissions: Set(root)
    15/04/30 08:24:40 INFO slf4j.Slf4jLogger: Slf4jLogger started
    15/04/30 08:24:40 INFO Remoting: Starting remoting
    15/04/30 08:24:40 INFO Remoting: Remoting started; listening on addresses :[akka.tcp://sparkDriver@karei.netact.noklab.net:41232]
    15/04/30 08:24:40 INFO util.Utils: Successfully started service 'sparkDriver' on port 41232.
    15/04/30 08:24:40 INFO spark.SparkEnv: Registering MapOutputTracker
    15/04/30 08:24:40 INFO spark.SparkEnv: Registering BlockManagerMaster
    15/04/30 08:24:40 INFO storage.DiskBlockManager: Created local directory at /tmp/spark-5713b972-849e-44c9-9e92-8cd0b4de7668/blockmgr-24d0ad17-fef5-4179-9018-39057c519c5b
    15/04/30 08:24:40 INFO storage.MemoryStore: MemoryStore started with capacity 265.0 MB
    15/04/30 08:24:40 INFO spark.HttpFileServer: HTTP File server directory is /tmp/spark-ea736e60-3575-49b7-b8ba-1b2f8587bac3/httpd-eb6af892-2e5c-4331-a510-d882af085db4
    15/04/30 08:24:40 INFO spark.HttpServer: Starting HTTP Server
    15/04/30 08:24:40 INFO server.Server: jetty-8.y.z-SNAPSHOT
    15/04/30 08:24:40 INFO server.AbstractConnector: Started SocketConnector@0.0.0.0:32977
    15/04/30 08:24:40 INFO util.Utils: Successfully started service 'HTTP file server' on port 32977.
    15/04/30 08:24:40 INFO spark.SparkEnv: Registering OutputCommitCoordinator
    15/04/30 08:24:40 INFO server.Server: jetty-8.y.z-SNAPSHOT
    15/04/30 08:24:40 INFO server.AbstractConnector: Started SelectChannelConnector@0.0.0.0:4040
    15/04/30 08:24:40 INFO util.Utils: Successfully started service 'SparkUI' on port 4040.
    15/04/30 08:24:40 INFO ui.SparkUI: Started SparkUI at http://karei.netact.noklab.net:4040
    15/04/30 08:24:40 INFO spark.SparkContext: Added JAR file:/usr/local/spark/mylib/spark-demo-1.0.jar at http://10.9.167.55:32977/jars/spark-demo-1.0.jar with timestamp 1430371480725
    15/04/30 08:24:40 INFO executor.Executor: Starting executor ID <driver> on host localhost
    15/04/30 08:24:40 INFO util.AkkaUtils: Connecting to HeartbeatReceiver: akka.tcp://sparkDriver@karei.netact.noklab.net:41232/user/HeartbeatReceiver
    15/04/30 08:24:40 INFO netty.NettyBlockTransferService: Server created on 34209
    15/04/30 08:24:40 INFO storage.BlockManagerMaster: Trying to register BlockManager
    15/04/30 08:24:40 INFO storage.BlockManagerMasterActor: Registering block manager localhost:34209 with 265.0 MB RAM, BlockManagerId(<driver>, localhost, 34209)
    15/04/30 08:24:40 INFO storage.BlockManagerMaster: Registered BlockManager
    15/04/30 08:24:41 WARN util.SizeEstimator: Failed to check whether UseCompressedOops is set; assuming yes
    15/04/30 08:24:41 INFO storage.MemoryStore: ensureFreeSpace(165175) called with curMem=0, maxMem=277877882
    15/04/30 08:24:41 INFO storage.MemoryStore: Block broadcast_0 stored as values in memory (estimated size 161.3 KB, free 264.8 MB)
    15/04/30 08:24:41 INFO storage.MemoryStore: ensureFreeSpace(25610) called with curMem=165175, maxMem=277877882
    15/04/30 08:24:41 INFO storage.MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 25.0 KB, free 264.8 MB)
    15/04/30 08:24:41 INFO storage.BlockManagerInfo: Added broadcast_0_piece0 in memory on localhost:34209 (size: 25.0 KB, free: 265.0 MB)
    15/04/30 08:24:41 INFO storage.BlockManagerMaster: Updated info of block broadcast_0_piece0
    15/04/30 08:24:41 INFO spark.SparkContext: Created broadcast 0 from textFile at SparkDemo.java:30
    15/04/30 08:24:41 INFO mapred.FileInputFormat: Total input paths to process : 1
    15/04/30 08:24:41 INFO Configuration.deprecation: mapred.tip.id is deprecated. Instead, use mapreduce.task.id
    15/04/30 08:24:41 INFO Configuration.deprecation: mapred.task.id is deprecated. Instead, use mapreduce.task.attempt.id
    15/04/30 08:24:41 INFO Configuration.deprecation: mapred.task.is.map is deprecated. Instead, use mapreduce.task.ismap
    15/04/30 08:24:41 INFO Configuration.deprecation: mapred.task.partition is deprecated. Instead, use mapreduce.task.partition
    15/04/30 08:24:41 INFO Configuration.deprecation: mapred.job.id is deprecated. Instead, use mapreduce.job.id
    15/04/30 08:24:41 INFO spark.SparkContext: Starting job: saveAsTextFile at SparkDemo.java:53
    15/04/30 08:24:41 INFO scheduler.DAGScheduler: Registering RDD 3 (mapToPair at SparkDemo.java:39)
    15/04/30 08:24:41 INFO scheduler.DAGScheduler: Got job 0 (saveAsTextFile at SparkDemo.java:53) with 1 output partitions (allowLocal=false)
    15/04/30 08:24:41 INFO scheduler.DAGScheduler: Final stage: Stage 1(saveAsTextFile at SparkDemo.java:53)
    15/04/30 08:24:41 INFO scheduler.DAGScheduler: Parents of final stage: List(Stage 0)
    15/04/30 08:24:41 INFO scheduler.DAGScheduler: Missing parents: List(Stage 0)
    15/04/30 08:24:41 INFO scheduler.DAGScheduler: Submitting Stage 0 (MapPartitionsRDD[3] at mapToPair at SparkDemo.java:39), which has no missing parents
    15/04/30 08:24:41 INFO storage.MemoryStore: ensureFreeSpace(4344) called with curMem=190785, maxMem=277877882
    15/04/30 08:24:41 INFO storage.MemoryStore: Block broadcast_1 stored as values in memory (estimated size 4.2 KB, free 264.8 MB)
    15/04/30 08:24:41 INFO storage.MemoryStore: ensureFreeSpace(3085) called with curMem=195129, maxMem=277877882
    15/04/30 08:24:41 INFO storage.MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 3.0 KB, free 264.8 MB)
    15/04/30 08:24:41 INFO storage.BlockManagerInfo: Added broadcast_1_piece0 in memory on localhost:34209 (size: 3.0 KB, free: 265.0 MB)
    15/04/30 08:24:41 INFO storage.BlockManagerMaster: Updated info of block broadcast_1_piece0
    15/04/30 08:24:41 INFO spark.SparkContext: Created broadcast 1 from broadcast at DAGScheduler.scala:839
    15/04/30 08:24:41 INFO scheduler.DAGScheduler: Submitting 1 missing tasks from Stage 0 (MapPartitionsRDD[3] at mapToPair at SparkDemo.java:39)
    15/04/30 08:24:41 INFO scheduler.TaskSchedulerImpl: Adding task set 0.0 with 1 tasks
    15/04/30 08:24:41 INFO scheduler.TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, localhost, PROCESS_LOCAL, 1343 bytes)
    15/04/30 08:24:41 INFO executor.Executor: Running task 0.0 in stage 0.0 (TID 0)
    15/04/30 08:24:41 INFO executor.Executor: Fetching http://10.9.167.55:32977/jars/spark-demo-1.0.jar with timestamp 1430371480725
    15/04/30 08:24:42 INFO util.Utils: Fetching http://10.9.167.55:32977/jars/spark-demo-1.0.jar to /tmp/spark-4cb44a60-1a2b-4490-bf6a-f123e6c13949/userFiles-22bb9d0d-6d44-48bd-910b-8a0b6581d11a/fetchFileTemp4845879217433172142.tmp
    15/04/30 08:24:42 INFO executor.Executor: Adding file:/tmp/spark-4cb44a60-1a2b-4490-bf6a-f123e6c13949/userFiles-22bb9d0d-6d44-48bd-910b-8a0b6581d11a/spark-demo-1.0.jar to class loader
    15/04/30 08:24:42 INFO rdd.HadoopRDD: Input split: file:/home/hadoop/input/test.log:0+119
    15/04/30 08:24:42 INFO executor.Executor: Finished task 0.0 in stage 0.0 (TID 0). 2003 bytes result sent to driver
    15/04/30 08:24:42 INFO scheduler.TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 274 ms on localhost (1/1)
    15/04/30 08:24:42 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Stage 0 (mapToPair at SparkDemo.java:39) finished in 0.293 s
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: looking for newly runnable stages
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: running: Set()
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: waiting: Set(Stage 1)
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: failed: Set()
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Missing parents for Stage 1: List()
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Submitting Stage 1 (MapPartitionsRDD[5] at saveAsTextFile at SparkDemo.java:53), which is now runnable
    15/04/30 08:24:42 INFO storage.MemoryStore: ensureFreeSpace(127584) called with curMem=198214, maxMem=277877882
    15/04/30 08:24:42 INFO storage.MemoryStore: Block broadcast_2 stored as values in memory (estimated size 124.6 KB, free 264.7 MB)
    15/04/30 08:24:42 INFO storage.MemoryStore: ensureFreeSpace(77104) called with curMem=325798, maxMem=277877882
    15/04/30 08:24:42 INFO storage.MemoryStore: Block broadcast_2_piece0 stored as bytes in memory (estimated size 75.3 KB, free 264.6 MB)
    15/04/30 08:24:42 INFO storage.BlockManagerInfo: Added broadcast_2_piece0 in memory on localhost:34209 (size: 75.3 KB, free: 264.9 MB)
    15/04/30 08:24:42 INFO storage.BlockManagerMaster: Updated info of block broadcast_2_piece0
    15/04/30 08:24:42 INFO spark.SparkContext: Created broadcast 2 from broadcast at DAGScheduler.scala:839
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Submitting 1 missing tasks from Stage 1 (MapPartitionsRDD[5] at saveAsTextFile at SparkDemo.java:53)
    15/04/30 08:24:42 INFO scheduler.TaskSchedulerImpl: Adding task set 1.0 with 1 tasks
    15/04/30 08:24:42 INFO scheduler.TaskSetManager: Starting task 0.0 in stage 1.0 (TID 1, localhost, PROCESS_LOCAL, 1114 bytes)
    15/04/30 08:24:42 INFO executor.Executor: Running task 0.0 in stage 1.0 (TID 1)
    15/04/30 08:24:42 INFO storage.ShuffleBlockFetcherIterator: Getting 1 non-empty blocks out of 1 blocks
    15/04/30 08:24:42 INFO storage.ShuffleBlockFetcherIterator: Started 0 remote fetches in 4 ms
    15/04/30 08:24:42 INFO output.FileOutputCommitter: Saved output of task 'attempt_201504300824_0001_m_000000_1' to file:/home/hadoop/output/_temporary/0/task_201504300824_0001_m_000000
    15/04/30 08:24:42 INFO mapred.SparkHadoopMapRedUtil: attempt_201504300824_0001_m_000000_1: Committed
    15/04/30 08:24:42 INFO executor.Executor: Finished task 0.0 in stage 1.0 (TID 1). 1828 bytes result sent to driver
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Stage 1 (saveAsTextFile at SparkDemo.java:53) finished in 0.171 s
    15/04/30 08:24:42 INFO scheduler.TaskSetManager: Finished task 0.0 in stage 1.0 (TID 1) in 171 ms on localhost (1/1)
    15/04/30 08:24:42 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 1.0, whose tasks have all completed, from pool
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Job 0 finished: saveAsTextFile at SparkDemo.java:53, took 0.640220 s
    15/04/30 08:24:42 INFO storage.MemoryStore: ensureFreeSpace(76899) called with curMem=402902, maxMem=277877882
    15/04/30 08:24:42 INFO storage.MemoryStore: Block broadcast_3 stored as values in memory (estimated size 75.1 KB, free 264.5 MB)
    15/04/30 08:24:42 INFO storage.MemoryStore: ensureFreeSpace(36686) called with curMem=479801, maxMem=277877882
    15/04/30 08:24:42 INFO storage.MemoryStore: Block broadcast_3_piece0 stored as bytes in memory (estimated size 35.8 KB, free 264.5 MB)
    15/04/30 08:24:42 INFO storage.BlockManagerInfo: Added broadcast_3_piece0 in memory on localhost:34209 (size: 35.8 KB, free: 264.9 MB)
    15/04/30 08:24:42 INFO storage.BlockManagerMaster: Updated info of block broadcast_3_piece0
    15/04/30 08:24:42 INFO spark.SparkContext: Created broadcast 3 from textFile at SparkDemo.java:58
    15/04/30 08:24:42 INFO mapred.FileInputFormat: Total input paths to process : 1
    15/04/30 08:24:42 INFO spark.SparkContext: Starting job: count at SparkDemo.java:67
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Got job 1 (count at SparkDemo.java:67) with 1 output partitions (allowLocal=false)
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Final stage: Stage 2(count at SparkDemo.java:67)
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Parents of final stage: List()
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Missing parents: List()
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Submitting Stage 2 (MapPartitionsRDD[8] at filter at SparkDemo.java:60), which has no missing parents
    15/04/30 08:24:42 INFO storage.MemoryStore: ensureFreeSpace(2936) called with curMem=516487, maxMem=277877882
    15/04/30 08:24:42 INFO storage.MemoryStore: Block broadcast_4 stored as values in memory (estimated size 2.9 KB, free 264.5 MB)
    15/04/30 08:24:42 INFO storage.MemoryStore: ensureFreeSpace(2102) called with curMem=519423, maxMem=277877882
    15/04/30 08:24:42 INFO storage.MemoryStore: Block broadcast_4_piece0 stored as bytes in memory (estimated size 2.1 KB, free 264.5 MB)
    15/04/30 08:24:42 INFO storage.BlockManagerInfo: Added broadcast_4_piece0 in memory on localhost:34209 (size: 2.1 KB, free: 264.9 MB)
    15/04/30 08:24:42 INFO storage.BlockManagerMaster: Updated info of block broadcast_4_piece0
    15/04/30 08:24:42 INFO spark.SparkContext: Created broadcast 4 from broadcast at DAGScheduler.scala:839
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Submitting 1 missing tasks from Stage 2 (MapPartitionsRDD[8] at filter at SparkDemo.java:60)
    15/04/30 08:24:42 INFO scheduler.TaskSchedulerImpl: Adding task set 2.0 with 1 tasks
    15/04/30 08:24:42 INFO scheduler.TaskSetManager: Starting task 0.0 in stage 2.0 (TID 2, localhost, PROCESS_LOCAL, 1354 bytes)
    15/04/30 08:24:42 INFO executor.Executor: Running task 0.0 in stage 2.0 (TID 2)
    15/04/30 08:24:42 INFO rdd.HadoopRDD: Input split: file:/home/hadoop/input/test.log:0+119
    15/04/30 08:24:42 INFO executor.Executor: Finished task 0.0 in stage 2.0 (TID 2). 1830 bytes result sent to driver
    15/04/30 08:24:42 INFO scheduler.TaskSetManager: Finished task 0.0 in stage 2.0 (TID 2) in 8 ms on localhost (1/1)
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Stage 2 (count at SparkDemo.java:67) finished in 0.008 s
    15/04/30 08:24:42 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 2.0, whose tasks have all completed, from pool
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Job 1 finished: count at SparkDemo.java:67, took 0.017870 s
    15/04/30 08:24:42 INFO spark.SparkContext: Starting job: count at SparkDemo.java:74
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Got job 2 (count at SparkDemo.java:74) with 1 output partitions (allowLocal=false)
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Final stage: Stage 3(count at SparkDemo.java:74)
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Parents of final stage: List()
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Missing parents: List()
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Submitting Stage 3 (MapPartitionsRDD[9] at filter at SparkDemo.java:69), which has no missing parents
    15/04/30 08:24:42 INFO storage.MemoryStore: ensureFreeSpace(3040) called with curMem=521525, maxMem=277877882
    15/04/30 08:24:42 INFO storage.MemoryStore: Block broadcast_5 stored as values in memory (estimated size 3.0 KB, free 264.5 MB)
    15/04/30 08:24:42 INFO storage.MemoryStore: ensureFreeSpace(2156) called with curMem=524565, maxMem=277877882
    15/04/30 08:24:42 INFO storage.MemoryStore: Block broadcast_5_piece0 stored as bytes in memory (estimated size 2.1 KB, free 264.5 MB)
    15/04/30 08:24:42 INFO storage.BlockManagerInfo: Added broadcast_5_piece0 in memory on localhost:34209 (size: 2.1 KB, free: 264.9 MB)
    15/04/30 08:24:42 INFO storage.BlockManagerMaster: Updated info of block broadcast_5_piece0
    15/04/30 08:24:42 INFO spark.SparkContext: Created broadcast 5 from broadcast at DAGScheduler.scala:839
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Submitting 1 missing tasks from Stage 3 (MapPartitionsRDD[9] at filter at SparkDemo.java:69)
    15/04/30 08:24:42 INFO scheduler.TaskSchedulerImpl: Adding task set 3.0 with 1 tasks
    15/04/30 08:24:42 INFO scheduler.TaskSetManager: Starting task 0.0 in stage 3.0 (TID 3, localhost, PROCESS_LOCAL, 1354 bytes)
    15/04/30 08:24:42 INFO executor.Executor: Running task 0.0 in stage 3.0 (TID 3)
    15/04/30 08:24:42 INFO rdd.HadoopRDD: Input split: file:/home/hadoop/input/test.log:0+119
    15/04/30 08:24:42 INFO executor.Executor: Finished task 0.0 in stage 3.0 (TID 3). 1830 bytes result sent to driver
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Stage 3 (count at SparkDemo.java:74) finished in 0.009 s
    15/04/30 08:24:42 INFO scheduler.TaskSetManager: Finished task 0.0 in stage 3.0 (TID 3) in 9 ms on localhost (1/1)
    15/04/30 08:24:42 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 3.0, whose tasks have all completed, from pool
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Job 2 finished: count at SparkDemo.java:74, took 0.019982 s
    15/04/30 08:24:42 INFO spark.SparkContext: Starting job: collect at SparkDemo.java:81
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Got job 3 (collect at SparkDemo.java:81) with 1 output partitions (allowLocal=false)
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Final stage: Stage 4(collect at SparkDemo.java:81)
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Parents of final stage: List()
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Missing parents: List()
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Submitting Stage 4 (MapPartitionsRDD[10] at filter at SparkDemo.java:76), which has no missing parents
    15/04/30 08:24:42 INFO storage.MemoryStore: ensureFreeSpace(3056) called with curMem=526721, maxMem=277877882
    15/04/30 08:24:42 INFO storage.MemoryStore: Block broadcast_6 stored as values in memory (estimated size 3.0 KB, free 264.5 MB)
    15/04/30 08:24:42 INFO storage.MemoryStore: ensureFreeSpace(2164) called with curMem=529777, maxMem=277877882
    15/04/30 08:24:42 INFO storage.MemoryStore: Block broadcast_6_piece0 stored as bytes in memory (estimated size 2.1 KB, free 264.5 MB)
    15/04/30 08:24:42 INFO storage.BlockManagerInfo: Added broadcast_6_piece0 in memory on localhost:34209 (size: 2.1 KB, free: 264.9 MB)
    15/04/30 08:24:42 INFO storage.BlockManagerMaster: Updated info of block broadcast_6_piece0
    15/04/30 08:24:42 INFO spark.SparkContext: Created broadcast 6 from broadcast at DAGScheduler.scala:839
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Submitting 1 missing tasks from Stage 4 (MapPartitionsRDD[10] at filter at SparkDemo.java:76)
    15/04/30 08:24:42 INFO scheduler.TaskSchedulerImpl: Adding task set 4.0 with 1 tasks
    15/04/30 08:24:42 INFO scheduler.TaskSetManager: Starting task 0.0 in stage 4.0 (TID 4, localhost, PROCESS_LOCAL, 1354 bytes)
    15/04/30 08:24:42 INFO executor.Executor: Running task 0.0 in stage 4.0 (TID 4)
    15/04/30 08:24:42 INFO rdd.HadoopRDD: Input split: file:/home/hadoop/input/test.log:0+119
    15/04/30 08:24:42 INFO executor.Executor: Finished task 0.0 in stage 4.0 (TID 4). 1828 bytes result sent to driver
    15/04/30 08:24:42 INFO scheduler.TaskSetManager: Finished task 0.0 in stage 4.0 (TID 4) in 7 ms on localhost (1/1)
    15/04/30 08:24:42 INFO scheduler.TaskSchedulerImpl: Removed TaskSet 4.0, whose tasks have all completed, from pool
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Stage 4 (collect at SparkDemo.java:81) finished in 0.006 s
    15/04/30 08:24:42 INFO scheduler.DAGScheduler: Job 3 finished: collect at SparkDemo.java:81, took 0.017157 s
    [root@karei spark]#

## 8. 输出结果 ##

[root@karei spark]# cat /home/hadoop/output/part-00000
    (Spark,2)
    (Operation,1)
    (ORACLE,1)
    (ERROR,1)
    (start,1)
    (INFO,2)
    (Configuration,1)
    (end,1)
    (Database,1)
    (testing,2)
    (HDFS,1)
    (WARN,1)
    (:,4)
    (Hadoop,1)
    [root@karei spark]#

[http_spark.apache.org_downloads.html]: http://spark.apache.org/downloads.html