Spark：java api实现word count统计

方案一：使用reduceByKey

数据word.txt

张三
李四
王五
李四
王五
李四
王五
李四
王五
王五
李四
李四
李四
李四
李四

代码：

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.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.rdd.RDD;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

public class HelloWord {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().master("local[*]").appName("Spark").getOrCreate();
        final JavaSparkContext ctx = JavaSparkContext.fromSparkContext(spark.sparkContext());

        RDD<String> rdd = spark.sparkContext().textFile("C:\Users\boco\Desktop\word.txt", 1);
        JavaRDD<String> javaRDD = rdd.toJavaRDD();

        JavaPairRDD<String, Integer> javaRDDMap = javaRDD.mapToPair(new PairFunction<String, String, Integer>() {
            public Tuple2<String, Integer> call(String s) {
                return new Tuple2<String, Integer>(s, 1);
            }
        });

        JavaPairRDD<String, Integer> result = javaRDDMap.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception {
                return integer + integer2;
            }
        });

        System.out.println(result.collect());
    }
}

输出：

[(张三,1), (李四,9), (王五,5)]

方案二：使用spark sql

使用spark sql实现代码：

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.util.ArrayList;

public class HelloWord {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().master("local[*]").appName("Spark").getOrCreate();
        final JavaSparkContext ctx = JavaSparkContext.fromSparkContext(spark.sparkContext());

        JavaRDD<Row> rows = spark.read().text("C:\Users\boco\Desktop\word.txt").toJavaRDD();

        ArrayList<StructField> fields = new ArrayList<StructField>();
        StructField field = null;
        field = DataTypes.createStructField("key", DataTypes.StringType, true);
        fields.add(field);

        StructType schema = DataTypes.createStructType(fields);

        Dataset<Row> ds = spark.createDataFrame(rows, schema);

        ds.createOrReplaceTempView("words");

        Dataset<Row> result = spark.sql("select key,count(0) as key_count from words group by key");

        result.show();
    }
}

结果：

+---+---------+
|key|key_count|
+---+---------+
| 王五|        5|
| 李四|        9|
| 张三|        1|
+---+---------+

方案二：使用spark streaming实时流分析

参考《http://spark.apache.org/docs/latest/streaming-programming-guide.html》

First, we create a JavaStreamingContext object, which is the main entry point for all streaming functionality. We create a local StreamingContext with two execution threads, and a batch interval of 1 second.

import org.apache.spark.*;
import org.apache.spark.api.java.function.*;
import org.apache.spark.streaming.*;
import org.apache.spark.streaming.api.java.*;
import scala.Tuple2;

// Create a local StreamingContext with two working thread and batch interval of 1 second
SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount");
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));

Using this context, we can create a DStream that represents streaming data from a TCP source, specified as hostname (e.g. localhost) and port (e.g. 9999).

// Create a DStream that will connect to hostname:port, like localhost:9999
JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 9999);

This lines DStream represents the stream of data that will be received from the data server. Each record in this stream is a line of text. Then, we want to split the lines by space into words.

// Split each line into words
JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());

flatMap is a DStream operation that creates a new DStream by generating multiple new records from each record in the source DStream. In this case, each line will be split into multiple words and the stream of words is represented as the words DStream. Note that we defined the transformation using a FlatMapFunction object. As we will discover along the way, there are a number of such convenience classes in the Java API that help defines DStream transformations.

Next, we want to count these words.

// Count each word in each batch
JavaPairDStream<String, Integer> pairs = words.mapToPair(s -> new Tuple2<>(s, 1));
JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey((i1, i2) -> i1 + i2);

// Print the first ten elements of each RDD generated in this DStream to the console
wordCounts.print();

The words DStream is further mapped (one-to-one transformation) to a DStream of (word, 1) pairs, using a PairFunction object. Then, it is reduced to get the frequency of words in each batch of data, using a Function2 object. Finally, wordCounts.print() will print a few of the counts generated every second.

Note that when these lines are executed, Spark Streaming only sets up the computation it will perform after it is started, and no real processing has started yet. To start the processing after all the transformations have been setup, we finally call start method.

jssc.start();              // Start the computation
jssc.awaitTermination();   // Wait for the computation to terminate

The complete code can be found in the Spark Streaming example JavaNetworkWordCount. 

If you have already downloaded and built Spark, you can run this example as follows. You will first need to run Netcat (a small utility found in most Unix-like systems) as a data server by using

$ nc -lk 9999

Then, in a different terminal, you can start the example by using

$ ./bin/run-example streaming.JavaNetworkWordCount localhost 9999

完整代码：

import java.util.Arrays;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;

import scala.Tuple2;

public class HelloWord {
    public static void main(String[] args) throws InterruptedException {
        // Create a local StreamingContext with two working thread and batch interval of
        // 1 second
        SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("NetworkWordCount");
        JavaSparkContext jsc=new JavaSparkContext(conf);
        jsc.setLogLevel("WARN");
        JavaStreamingContext jssc = new JavaStreamingContext(jsc, Durations.seconds(60));
        
        // Create a DStream that will connect to hostname:port, like localhost:9999
        JavaReceiverInputDStream<String> lines = jssc.socketTextStream("xx.xx.xx.xx", 19999);

        // Split each line into words
        JavaDStream<String> words = lines.flatMap(x -> Arrays.asList(x.split(" ")).iterator());

        // Count each word in each batch
        JavaPairDStream<String, Integer> pairs = words.mapToPair(s -> new Tuple2<>(s, 1));
        JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey((i1, i2) -> i1 + i2);

        // Print the first ten elements of each RDD generated in this DStream to the
        // console
        wordCounts.print();

        jssc.start(); // Start the computation
        jssc.awaitTermination(); // Wait for the computation to terminate
    }
}

测试：

[root@abced dx]# nc -lk 19999
hellow wrd
hello word
hello word
hello dkk
hl
hello
hello
hello word
hello word
hello java
hello c@
hello hadoop]
hello spark
hello word
hello kafka
hello c
hello c#
hello .net core
net cre
workd
hle
hello words
hke hjh
hek 23
hel 23
hl3 323
hhk 68
hke 84

程序执行结果：

-------------------------------------------
Time: 1533781920000 ms
-------------------------------------------
(c,1)
(spark,1)
(kafka,1)
(c#,1)
(hello,9)
(java,1)
(c@,1)
(hadoop],1)
(word,2)

18/08/09 10:32:05 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
18/08/09 10:32:05 WARN BlockManager: Block input-0-1533781925200 replicated to only 0 peer(s) instead of 1 peers
18/08/09 10:32:08 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
18/08/09 10:32:08 WARN BlockManager: Block input-0-1533781928000 replicated to only 0 peer(s) instead of 1 peers
18/08/09 10:32:11 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
18/08/09 10:32:11 WARN BlockManager: Block input-0-1533781931200 replicated to only 0 peer(s) instead of 1 peers
18/08/09 10:32:14 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
18/08/09 10:32:14 WARN BlockManager: Block input-0-1533781934600 replicated to only 0 peer(s) instead of 1 peers
-------------------------------------------
Time: 1533781980000 ms
-------------------------------------------
(hle,1)
(words,1)
(.net,1)
(hello,2)
(workd,1)
(cre,1)
(net,1)
(core,1)

18/08/09 10:33:08 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
18/08/09 10:33:08 WARN BlockManager: Block input-0-1533781988000 replicated to only 0 peer(s) instead of 1 peers
18/08/09 10:33:11 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
18/08/09 10:33:11 WARN BlockManager: Block input-0-1533781991000 replicated to only 0 peer(s) instead of 1 peers
18/08/09 10:33:14 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
18/08/09 10:33:14 WARN BlockManager: Block input-0-1533781994200 replicated to only 0 peer(s) instead of 1 peers
18/08/09 10:33:17 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
18/08/09 10:33:17 WARN BlockManager: Block input-0-1533781997400 replicated to only 0 peer(s) instead of 1 peers
18/08/09 10:33:20 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
18/08/09 10:33:20 WARN BlockManager: Block input-0-1533782000400 replicated to only 0 peer(s) instead of 1 peers
18/08/09 10:33:25 WARN RandomBlockReplicationPolicy: Expecting 1 replicas with only 0 peer/s.
18/08/09 10:33:25 WARN BlockManager: Block input-0-1533782005600 replicated to only 0 peer(s) instead of 1 peers
-------------------------------------------
Time: 1533782040000 ms
-------------------------------------------
(68,1)
(hhk,1)
(hek,1)
(hel,1)
(84,1)
(hjh,1)
(23,2)
(hke,2)
(323,1)
(hl3,1)

结论：是一批一批的处理的，不进行累加，每一批统计并不是累加之前的数据，而是针对当前接收到这一批数据的处理。