/**
 *
 *  Spark standalone or Mesos with cluster deploy mode only:
 *  在提交application的时候  添加 --supervise 选项  如果Driver挂掉 会自动启动一个Driver
 *  	SparkStreaming
 * @author Jeffrey.deng
 *
 */
public class SparkStreamingOnHDFS {
	public static void main(String[] args) {
		final SparkConf conf = new SparkConf().setMaster("local[1]").setAppName("SparkStreamingOnHDFS");
		
		final String checkpointDirectory = "hdfs://hadoop1:9000/library/SparkStreaming/CheckPoint_20170523";
		
		JavaStreamingContextFactory factory = new JavaStreamingContextFactory() {
			@Override
			public JavaStreamingContext create() {  
				return createContext(checkpointDirectory,conf);
			}
		};
		
		// 通过factory来创建JavaStreamingContext
		JavaStreamingContext jsc = JavaStreamingContext.getOrCreate(checkpointDirectory, factory); // 此路径下无driver信息则新建，有则根据信息恢复
		jsc.start();
		jsc.awaitTermination();
		// jsc.close();
	}

	@SuppressWarnings("deprecation")
	private static JavaStreamingContext createContext(String checkpointDirectory,SparkConf conf) {

		// If you do not see this printed,
		// that means the StreamingContext has been loaded from the new checkpoint
		System.out.println("Creating new context");
		SparkConf sparkConf = conf;
		// Create the context with a 1 second batch size

		JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, Durations.seconds(15));

		// 指定路径  会先恢复Driver需要的信息写到此路径下
		ssc.checkpoint(checkpointDirectory);
		
		/**
		 * 只是监控文件夹下新增的文件，减少的文件是监控不到的  文件内容有改动也是监控不到
		 */
		JavaDStream<String> lines = ssc.textFileStream("hdfs://hadoop1:9000/hdfs/");
		 
		JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

			@Override
			public Iterable<String> call(String s) {
				return Arrays.asList(s.split(" "));
			}
		});
		

		JavaPairDStream<String, Integer> ones = words.mapToPair(new PairFunction<String, String, Integer>() {
			@Override
			public Tuple2<String, Integer> call(String s) {
				return new Tuple2<String, Integer>(s, 1);
			}
		});

		JavaPairDStream<String, Integer> counts = ones.reduceByKey(new Function2<Integer, Integer, Integer>() {
			@Override
			public Integer call(Integer i1, Integer i2) {
				return i1 + i2;
			}
		});
	 
		counts.print();
		return ssc;
	}

}

第一次启动程序成功后，会输出

        System.out.println("Creating new context");