Apache Spark - WordCount 예제 (JAVA)

package practice;



import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HBaseAdmin;
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 scala.Tuple2;

import java.util.Arrays;

public class WordCount {

    // args[] = { masterInfo, inputTextPath, outputTextPath }
    public static void main(String[] args) {
        if(ArrayUtils.getLength(args) != 3) {
            System.out.println("check args");
            return;
        }

        String appName = "WordCountExample";
        JavaSparkContext sc = new JavaSparkContext(new SparkConf().setAppName(appName).setMaster(args[0]));

        try {
            JavaRDD<String> inputRDD = sc.textFile(args[1]);
            JavaPairRDD<String, Integer> resultRDD = process(inputRDD);
            handleResult(resultRDD, args[2]);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //spark 연결 종료
            sc.stop();
        }
    }

    public static JavaPairRDD<String, Integer> process(JavaRDD<String> inputRDD) {
        JavaRDD<String> words = inputRDD.flatMap((String s)-> Arrays.asList(s.split(" ")).iterator());
        JavaPairRDD<String, Integer> wcPair = words.mapToPair((String w)-> new Tuple2<>(w,1));
        JavaPairRDD<String, Integer> result = wcPair.reduceByKey((Integer c1, Integer c2) -> c1 + c2);
        return result;
    }

    public static void handleResult(JavaPairRDD<String, Integer> resultRDD, String output) {
        resultRDD.saveAsTextFile(output);
    }
}

 

// 실행방법

 

<spark_home_dir>/bin/spark-submit \

-- class com.wikibooks.spark.ch1.WordCount \

<jar 파일경로>/beginning-spark-examples.jar \

local[*] \

<spark_home_dir>/README.md \

<spark_home_dir>/testresult