/*

  * Licensed to the Apache Software Foundation (ASF) under one or more

  * contributor license agreements.  See the NOTICE file distributed with

  * this work for additional information regarding copyright ownership.

  * The ASF licenses this file to You under the Apache License, Version 2.0

  * (the "License"); you may not use this file except in compliance with

  * the License.  You may obtain a copy of the License at

  *

  *    http://www.apache.org/licenses/LICENSE-2.0

  *

  * Unless required by applicable law or agreed to in writing, software

  * distributed under the License is distributed on an "AS IS" BASIS,

  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

  * See the License for the specific language governing permissions and

  * limitations under the License.

  */  

 package com.east.spark.stream;  

 import java.util.Arrays;

 import java.util.List;

 import java.util.regex.Pattern;  

 import org.apache.spark.api.java.JavaPairRDD;

 import org.apache.spark.api.java.JavaRDD;

 import org.apache.spark.sql.SparkSession;  

 import scala.Tuple2;  

 public final class JavaWordCount2 {

     private static final Pattern SPACE = Pattern.compile(" ");  

     public static void main(String[] args) throws Exception {  

         args = new String[] { "D:/tmp/spark/test.txt" };  

         if (args.length < 1) {

             System.err.println("Usage: JavaWordCount <file>");

             System.exit(1);

         }  

         SparkSession spark = SparkSession.builder().appName("JavaWordCount").master("local").getOrCreate();  

         // SparkConf conf = new

         // SparkConf().setAppName("ingini-spark-java8").setMaster("local");  

         JavaRDD<String> lines = spark.read().textFile(args[0]).javaRDD();  

         JavaRDD<String> words = lines.flatMap(line -> Arrays.asList(line.split(" ")).iterator());  

         JavaPairRDD<String, Integer> counts = words.mapToPair(w -> new Tuple2<String, Integer>(w, 1))

                 .reduceByKey((x, y) -> x + y);

         // counts.collect();  

         List<Tuple2<String, Integer>> output = counts.collect();

         for (Tuple2<?, ?> tuple : output) {

             System.out.println(tuple._1() + ":== " + tuple._2());

         }  

         spark.stop();

     }

 }  

 JavaRDD<String> lines = sc.textFile("src/main/resources/a.txt");

         JavaPairRDD<String, Integer> counts = lines.flatMap(line -> Arrays.asList(line.split(REGEX)))

                 .mapToPair(word -> new Tuple2(word, 1))

                 .reduceByKey((x, y) -> (Integer) x + (Integer) y)

                 .sortByKey();

         counts.foreach(stringIntegerTuple2 ->System.out.println( stringIntegerTuple2._1+":"+stringIntegerTuple2._2));