public static void main(String[] args) {
		SparkConf conf = new SparkConf()
				.setMaster("local[2]")
				.setAppName("UpdateStateByKeyWordCount");  
	JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));

	jssc.checkpoint("hdfs://hadoop01:9000/wordcount_checkpoint");  
	// 命令： nc -lk 9999 
	JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 9999);
		
	JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String, String>() {

			private static final long serialVersionUID = 1L;

			@Override
			public Iterable<String> call(String line) throws Exception {
				return Arrays.asList(line.split(" "));  
			}
			
		});
		
		JavaPairDStream<String, Integer> pairs = words.mapToPair(
				
				new PairFunction<String, String, Integer>() {

					private static final long serialVersionUID = 1L;

					@Override
					public Tuple2<String, Integer> call(String word)
							throws Exception {
						return new Tuple2<String, Integer>(word, 1);
					}
					
				});

		JavaPairDStream<String, Integer> wordCounts = pairs.updateStateByKey(
				
		// 這裡的Optional，相當於Scala中的樣例類，就是Option，可以這麼理解
		// 它代表了一個值的存在狀態，可能存在，也可能不存在
		new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() {

			private static final long serialVersionUID = 1L;

			@Override
			public Optional<Integer> call(List<Integer> values,
					Optional<Integer> state) throws Exception {
				Integer newValue = 0;
				if(state.isPresent()) {
					newValue = state.get();
				}
				for(Integer value : values) {
					newValue += value;
				}
				
				return Optional.of(newValue);  
			}
			
		});
		
		wordCounts.print();
		
		jssc.start();
		jssc.awaitTermination();
		jssc.close();
	}
 

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
        .setMaster("local[2]")  
        .setAppName("UpdateStateByKeyWordCount")
    val ssc = new StreamingContext(conf, Seconds(5))
    ssc.checkpoint("hdfs://spark1:9000/wordcount_checkpoint")  
    
    val lines = ssc.socketTextStream("spark1", 9999)
    val words = lines.flatMap { _.split(" ") }   
    val pairs = words.map { word => (word, 1) } 
    val wordCounts = pairs.updateStateByKey((values: Seq[Int], state: Option[Int]) => {
      var newValue = state.getOrElse(0)    
      for(value <- values) {
        newValue += value
      }
      Option(newValue)  
    })
    
    wordCounts.print()  
    
    ssc.start()
    ssc.awaitTermination()
  }