Java版本

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.function.FlatMapFunction;

import org.apache.spark.api.java.function.Function2;

import org.apache.spark.api.java.function.PairFunction;

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;

import java.util.Iterator;

/**

 * 实时wordcount程序

 */

public class JavaSparkStreaming {

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

        //创建SparkConf对象

        //要给它设置一个Master属性，但是我们测试的时候使用local模式

        //local后面必须跟一个方括号，里面填写一个数字，数字代表了我们用几个线程来执行

        //我们的spark streaming程序

        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("JavaSparkStreaming");

        //创建JavaStreamingContext对象

        //类似于Spark Core中的JavaSparkContext，类似于Spark SQL中的SQLContext

        //该对象除了接收SparkConf对象外

        //还必须接收一个batch interval参数，就是说，每手机多长时间的数据，划分为一个batch进行处理

        //当前设置为1秒

        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(1));

        //首先，创建输入DStream，代表了一个从数据源（比如kafka、socket）来的持续不断的实时数据流

        //调用JavaStreamingContext的socketTextStream()方法，创建一个数据源为socket网络端口的数据流

        //JavaReceiverInputDStream代表了一个输入的DStream

        //socketTextStream()方法接收两个基本参数，第一个是监听那个主机上的端口，第二个是监听哪个端口

        JavaReceiverInputDStream<String> lines = jssc.socketTextStream("localhost", 9999);

        //至此，可以理解为JavaReceiverInputDStream中的，每隔一秒，会有一个RDD

        //其中封装了这一秒发送过来的数据

        //RDD的元素类型为String，即一行一行的文本

        //所以，这里JavaReceiverInputDStream的泛型类型为<String>，其实代表了底层的RDD的泛型类型

        //开始对接收到的数据，执行计算，使用Spark Core提供的算子，执行应用在DStream中即可

        //在底层，实际上是会对DStream中的一个一个的RDD，执行我们应用在DStream上的算子

        //产生新的RDD，会作为新DStream中的RDD

        JavaDStream<String> words = lines.flatMap(

                new FlatMapFunction<String, String>() {

                    @Override

                    public Iterator<String> call(String s) throws Exception {

                        return null;

                        //return Arrays.asList(line.spilt(" "));

                    }

                }

        );

        //此时，每秒的数据，一行一行的文本，就会被拆分为多个单词，words DStream中的RDD的元素类型

        //即为一个一个的单词

        //接着，开始进行flatMap、reduceByKey操作

        JavaPairDStream<String, Integer> pairs = words.mapToPair(

                new PairFunction<String, String, Integer>() {

                    @Override

                    public Tuple2<String, Integer> call(String word) throws Exception {

                        return new Tuple2<String, Integer>(word, 1);

                    }

                }

        );

        //用Java Streaming开发程序和Java Core很像

        //唯一不同的是，Spark Core中的JavaRDD、JavaPairRDD，都变成了JavaDStream、JavaPairDStream

        JavaPairDStream<String, Integer> wordCounts = pairs.reduceByKey(

                new Function2<Integer, Integer, Integer>() {

                    @Override

                    public Integer call(Integer v1, Integer v2) throws Exception {

                        return v1 + v2;

                    }

                }

        );

        //每秒中发送到指定socket端口上的数据，都会被line DStream接收到

        //然后lines DStream会把每秒的数据，也就是一行一行的文本，诸如hello world，封装为一个RDD

        //然后，就会对每秒中对应的RDD，执行后续的一系列算子操作

        //比如，对Lines RDD执行了flatMap之后，得到了一个words RDD，作为words DStream中的一个RDD

        //以此类推，直到生成最后一个wordCount RDD，作为wordCounts DStream中的一个RDD

        //此时，就得到了每秒钟发送过来的数据的单词统计

        //注意：Spark Streaming的计算模型，决定了我们必须自己来进行中间缓存的控制

        //比如写入redis等缓存

        //它的计算模型跟storm是完全不同的，storm是自己编写的一个一个的程序，运行在节点上

        //相当于一个一个的对象，可以自己在对象中控制缓存

        //但是Spark本身是函数式编程的计算模型，所以，比如在words或pairs DStream中

        //没法实例变量进行缓存

        //此时，就只能将最后计算出的wordCounts中的一个一个的RDD，写入外部的缓存，或者持久化DB

        //最后，每次计算完，都打印一下这一秒钟的单词技术情况

        //并休眠5秒钟，以便于我们测试和观察

        Thread.sleep(5000);

        wordCounts.print();

        //对JavaStreamingContext进行后续处理

        //必须调用JavaStreamingContext的start()方法，整个Spark Steaming Application才会启动执行

        //否则不会执行

        jssc.start();

        jssc.awaitTermination();

        jssc.close();

    }

}

Scala版本

import org.apache.spark.SparkConf

import org.apache.spark.streaming.{Seconds, StreamingContext}

/**

  * wordcount

  */

object ScalaSparkStreaming {

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf().setMaster("local[2]").setMaster("ScalaSparkStreaming")

    //scala中，创建的是StreamingContext

    val ssc = new StreamingContext(conf, Seconds(1))

    val lines = ssc.socketTextStream("localhost", 9999)

    val words = lines.flatMap {

      _.split(" ")

    }

    val pairs = words.map {

      word => (word, 1)

    }

    val wordCounts = pairs.reduceByKey {

      _ + _

    }

    Thread.sleep(5000)

    wordCounts.print()

    ssc.start()

    ssc.awaitTermination()

  }

}

pom.xml

切记要对应好Spark 和 Scala的版本。

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0"

         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>

    <groupId>spark</groupId>

    <artifactId>com.spark</artifactId>

    <version>1.0-SNAPSHOT</version>

    <name>SparkTest</name>

    <url>http://maven.apache.org</url>

    <properties>

        <maven.compiler.source>1.8</maven.compiler.source>

        <maven.compiler.target>1.8</maven.compiler.target>

        <spark.version>2.3.0</spark.version>

        <hadoop.version>2.6.4</hadoop.version>

        <encoding>UTF-8</encoding>

    </properties>

    <dependencies>

        <!-- spark依赖-->

        <dependency>

            <groupId>org.apache.spark</groupId>

            <artifactId>spark-core_2.11</artifactId>

            <version>${spark.version}</version>

        </dependency>

        <dependency>

            <groupId>org.apache.spark</groupId>

            <artifactId>spark-sql_2.11</artifactId>

            <version></version>

        </dependency>

        <dependency>

            <groupId>org.apache.spark</groupId>

            <artifactId>spark-streaming_2.11</artifactId>

            <version>${spark.version}</version>

        </dependency>

        <!-- hadoop依赖-->

        <dependency>

            <groupId>org.apache.hadoop</groupId>

            <artifactId>hadoop-client</artifactId>

            <version>${hadoop.version}</version>

        </dependency>

    </dependencies>

    <build>

        <pluginManagement>

            <plugins>

                <!-- 编译java的插件 -->

                <plugin>

                    <groupId>org.apache.maven.plugins</groupId>

                    <artifactId>maven-compiler-plugin</artifactId>

                    <version>3.5.1</version>

                </plugin>

            </plugins>

        </pluginManagement>

        <plugins>

            <plugin>

                <groupId>org.apache.maven.plugins</groupId>

                <artifactId>maven-compiler-plugin</artifactId>

                <executions>

                    <execution>

                        <phase>compile</phase>

                        <goals>

                            <goal>compile</goal>

                        </goals>

                    </execution>

                </executions>

            </plugin>

            <!-- 打包插件 -->

            <plugin>

                <groupId>org.apache.maven.plugins</groupId>

                <artifactId>maven-shade-plugin</artifactId>

                <version>2.4.3</version>

                <executions>

                    <execution>

                        <phase>package</phase>

                        <goals>

                            <goal>shade</goal>

                        </goals>

                        <configuration>

                            <filters>

                                <filter>

                                    <artifact>*:*</artifact>

                                    <excludes>

                                        <exclude>META-INF/*.SF</exclude>

                                        <exclude>META-INF/*.DSA</exclude>

                                        <exclude>META-INF/*.RSA</exclude>

                                    </excludes>

                                </filter>

                            </filters>

                        </configuration>

                    </execution>

                </executions>

            </plugin>

        </plugins>

    </build>

</project>