devbox
正经夜光杯
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

SparkStreaming的案例及应用_sparkstreaming案例

作者：正经夜光杯 | 2024-08-08 01:53:58

赞

踩

sparkstreaming案例

1.案例：通过网络监听端口的方式，实现SparkStreaming的单词计数功能,弊端就是不能全局累加，只能累加同一批的数据

创建Maven项目：

<?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>com.aura.spark</groupId>
    <artifactId>1711spark</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <scala.version>2.11.8</scala.version>
        <spark.version>2.3.0</spark.version>
        <hadoop.version>2.7.5</hadoop.version>
        <encoding>UTF-8</encoding>
    </properties>

    <dependencies>

        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>

        <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>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-graphx_2.11 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-graphx_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-mllib_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-hive -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.11</artifactId>
            <version>2.3.0</version>
        </dependency>




        <!-- https://mvnrepository.com/artifact/mysql/mysql-connector-java -->
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.6</version>
        </dependency>

        <!-- https://mvnrepository.com/artifact/org.apache.spark/spark-streaming-kafka-0-8_2.11 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-8_2.11</artifactId>
            <version>${spark.version}</version>

        </dependency>




    </dependencies>

    <build>
        <pluginManagement>
            <plugins>
                <plugin>
                    <groupId>net.alchim31.maven</groupId>
                    <artifactId>scala-maven-plugin</artifactId>
                    <version>3.2.2</version>
                </plugin>
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.5.1</version>
                </plugin>
            </plugins>
        </pluginManagement>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <executions>
                    <execution>
                        <id>scala-compile-first</id>
                        <phase>process-resources</phase>
                        <goals>
                            <goal>add-source</goal>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                    <execution>
                        <id>scala-test-compile</id>
                        <phase>process-test-resources</phase>
                        <goals>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

            <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>

创建Scala代码 NetWordCount.scala  (注意是object)

千万注意local[2]最少是2，否则什么事也不会做，因为有Recevier 启动起来就是一个task任务，就需要一个线程，只写1个线程（去接收数据）就没有线程处理数据了，所以在本地最少要写2

而且代码是微批处理并不是非常准确的实时计算，每Seconds（2）两秒运行一次本批数据

  1. import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
  2. import org.apache.spark.{SparkConf, SparkContext}
  3. import org.apache.spark.streaming.{Seconds, StreamingContext}
  4.  
  5. /**
  6.   * Create NetWordCount.scala by jenrey on 2018/5/11 21:58
  7.   */
  8. object NetWordCount {
  9.   def main(args: Array[String]): Unit = {
  10.     /**
  11.       * 初始化程序入口
  12.       */
  13.     val conf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("NetWordCount")
  14.     val sc = new SparkContext(conf)
  15.     val ssc = new StreamingContext(sc,Seconds(2))
  16.  
  17.     /*StreamingContext源码会createNewSparkContext,所以可以省略创建SparkContext对象。
  18.     val ssc = new StreamingContext(conf,Seconds(2))*/
  19.  
  20.     /**
  21.       * 通过程序入口获取DStream
  22.       * 我们通过监听的方式获取数据，监听主机名和端口。只要一有数据就可以获取到，相当于通过Socket的方式
  23.       */
  24.       //ReceiverInputDStream就是个DStream，继承InputDStream继承DStream(就是一个抽象类,其实就是个HashMap(Time，RDD[T])一个时间点对应一个RDD )
  25.     val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop04", 9999)
  26.  
  27.     /**
  28.       * 对DStream流进行操作
  29.       */
  30.       //下面都是Transformation操作
  31.     val wordCountDStream: DStream[(String, Int)] = dstream.flatMap(line => line.split(","))
  32.       .map((_, 1))
  33.       .reduceByKey(_ + _)
  34.     //output的操作类似于RDD的Action
  35.     wordCountDStream.print()  //把数据打印出来
  36.     /**
  37.       * 启动应用程序（固定操作）
  38.       */
  39.     //启动我们的程序
  40.     ssc.start();
  41.     //等待结束
  42.     ssc.awaitTermination();
  43.     //如果结束就释放资源
  44.     ssc.stop();
  45.   }
  46. }

在hadoop04几点上使用下面的命令：

[hadoop@hadoop04 ~]$ nc -lk 9999

================================================

（安装nc：）

[hadoop@hadoop04 ~]$ sudo yum install nc


输入nc命令查看是否安装成功


==================================================

先让代码跑起来。

再在hadoop04 的 nc下输入hadoop,hadoop


IDEA控制台的输出结果：



案例2：在HDFS的WordCount程序

  1. import org.apache.spark.streaming.dstream.DStream
  2. import org.apache.spark.{SparkConf, SparkContext}
  3. import org.apache.spark.streaming.{Seconds, StreamingContext}
  4.  
  5. /**
  6.   * 用SparkStreaming在HA模式下的HDFS跑WordCount程序
  7.   */
  8. object HDFSWordCount {
  9.  
  10.   def main(args: Array[String]): Unit = {
  11.     val conf = new SparkConf().setMaster("local[2]").setAppName("HDFSWordCount")
  12.     val sc = new SparkContext(conf)
  13.     val ssc = new StreamingContext(sc,Seconds(2))
  14.     /**
  15.       * 数据的输入
  16.       */
  17.       //监控的是一个目录即文件夹，新增文件就可以接收到了
  18.     val fileDStream: DStream[String] = ssc.textFileStream("hdfs://myha01/streaming")
  19.     /**
  20.       * 数据的处理
  21.       */
  22.     val wordCountDStream: DStream[(String, Int)] = fileDStream.flatMap(_.split(","))
  23.       .map((_, 1))
  24.       .reduceByKey(_ + _)
  25.  
  26.     /**
  27.       * 数据的输出
  28.       */
  29.     wordCountDStream.print()
  30.  
  31.     ssc.start()
  32.     ssc.awaitTermination()
  33.     ssc.stop()
  34.   }
  35.  
  36. }


1）在hdfs上新建一个文件夹用来监听文件夹，一旦往监听的文件夹下面放入文件就能监听的到数据

注意：如果用的是HA模式，一定要把两个配置（core-site.xml和hdfs-site.xml）文件放到resources下面


在hdfs创建一个空的用来测试代码的文件夹streaming

[hadoop@hadoop04 ~]$ hadoop fs -mkdir /streaming

把IDEA里面的代码跑起来！！！！

然后vim hello.txt写入

you,jump

i,jump

然后上传到streaming文件夹下面

[hadoop@hadoop04 ~]$ hadoop fs -put hello.txt /streaming




案例3：updateStateByKey的WordCount程序

  1. import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
  2. import org.apache.spark.{SparkConf, SparkContext}
  3. import org.apache.spark.streaming.{Seconds, StreamingContext}
  4.  
  5. /**
  6.   * Create by jenrey on 2018/5/13 15:09
  7.   */
  8. object UpdateStateByKeyWordCount {
  9.   def main(args: Array[String]): Unit = {
  10.     /**
  11.       * 初始化程序入口
  12.       */
  13.     val conf: SparkConf = new SparkConf().setMaster("local[2]").setAppName("UpdateStateByKeyWordCount")
  14.     val sc = new SparkContext(conf)
  15.     val ssc = new StreamingContext(sc, Seconds(2))
  16.     //注意一定要设置checkpoint目录，否则程序报错,但是这个HDFS目录一定要有权限，这个目录不用提前创建，自动创建
  17.     ssc.checkpoint("hdfs://myha01/StreamingCheckPoint")
  18.     /**
  19.       * 数据的输入
  20.       */
  21.     val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop04", 9999)
  22.     /**
  23.       * 数据的处理
  24.       */
  25.     val wordCountDStream: DStream[(String, Int)] = dstream.flatMap(_.split(","))
  26.       .map((_, 1))
  27.       //updateStateByKey(updateFunc: (Seq[V], Option[S]) => Option[S]) 注意里面是一个函数
  28.       //Option：Some：有值，None：没值
  29.       //ByKey:操作就是分组
  30.       //you,1
  31.       //you,1     => you,{1,1}和jump,{1}
  32.       //jump,1
  33.       //下面这个函数每一个key都会调用一次这个函数
  34.       //所以values:Seq[Int]代表List{1,1}  state:Option[Int]代表上一次这个单词出现了多少次，如果上一次没出现过就是None，如果出现过就是Some该1次就1次该2次就2次
  35.       .updateStateByKey((values: Seq[Int], state: Option[Int]) => {
  36.       val current: Int = values.sum
  37.       //上一次出现多少次如果有就有，没有就是0
  38.       val lastCount: Int = state.getOrElse(0)
  39.       //既然这个单词能调用这个方法，那么这个单词必然最少出现了一次就是Some，所以当前+上一次的就是这个单词一共出现多少次
  40.       Some(current + lastCount)
  41.     })
  42.  
  43.     /**
  44.       * 数据的输出
  45.       */
  46.     wordCountDStream.print()
  47.  
  48.     ssc.start()
  49.     ssc.awaitTermination()
  50.     ssc.stop()
  51.   }
  52. }

运行代码：

注意容易报错的两个点：

1.没有设置checkpoint导致报错

2.没有权限，报错，如下图


需要去设置权限

[hadoop@hadoop04 ~]$ hdfs dfs -chmod 777 /

然后运行代码：

在hadoop04节点进行发送数据：

[hadoop@hadoop04 ~]$ nc -lk 9999




案例4：程序运行后停止再运行接着上次的结果继续计算的WordCount程序

普通的为什么就不能停止后接着运行呢？因为又一次创建了程序入口（new StreamingContext）,是完全两个不同的程序入口，说白了就是关心Driver服务。两个Driver服务不同。

checkpoint可以把Driver服务里面的信息存到这个目录里，那么下一次我们启动的时候通过checkpoint里面的数据把Driver服务恢复成跟上一次一样的，那么再计算的时候就相当于对上一次的结果进行累加了

  1. import org.apache.spark.streaming.dstream.ReceiverInputDStream
  2. import org.apache.spark.{SparkConf, SparkContext}
  3. import org.apache.spark.streaming.{Seconds, StreamingContext}
  4.  
  5. object DriverHAWordCount {
  6.   def main(args: Array[String]): Unit = {
  7.     //注意本程序需要执行一次输入点数据，然后关闭再次执行就可以接着上次进行累加了
  8.     val checkpointDirectory: String = "hdfs://myha01/StreamingCheckPoint3";
  9.  
  10.     def functionToCreateContext(): StreamingContext = {
  11.       val conf = new SparkConf().setMaster("local[2]").setAppName("DriverHAWordCount")
  12.       val sc = new SparkContext(conf)
  13.       val ssc = new StreamingContext(sc, Seconds(2))
  14.       ssc.checkpoint(checkpointDirectory)
  15.       val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop04", 9999)
  16.       val wordCountDStream = dstream.flatMap(_.split(","))
  17.         .map((_, 1))
  18.         .updateStateByKey((values: Seq[Int], state: Option[Int]) => {
  19.           val currentCount = values.sum
  20.           val lastCount = state.getOrElse(0)
  21.           Some(currentCount + lastCount)
  22.         })
  23.  
  24.       wordCountDStream.print()
  25.  
  26.       ssc.start()
  27.       ssc.awaitTermination()
  28.       ssc.stop()
  29.       //最后一行代码就是返回的
  30.       ssc
  31.     }
  32.  
  33.     //从里面获取一个程序入口，如果checkpointDirectory目录里面有程序入口就用这个，如果没有就新new一个程序入口（或者说一个Driver服务）
  34.     val ssc = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _)
  35.  
  36.     ssc.start()
  37.     ssc.awaitTermination()
  38.     ssc.stop()
  39.  
  40.   }
  41.  
  42. }

运行程序：


关闭代码程序，再次运行，等待一会再输入数据




案例5：单词黑名单

  1. import org.apache.spark.broadcast.Broadcast
  2. import org.apache.spark.rdd.RDD
  3. import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
  4. import org.apache.spark.{SparkConf, SparkContext}
  5. import org.apache.spark.streaming.{Seconds, StreamingContext}
  6.  
  7. object WordBlack {
  8.   def main(args: Array[String]): Unit = {
  9.     val conf = new SparkConf().setMaster("local[2]").setAppName("WordBlack")
  10.     val sc: SparkContext = new SparkContext(conf)
  11.     val ssc = new StreamingContext(sc, Seconds(2))
  12.     /**
  13.       * 数据的输入
  14.       */
  15.     val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop04", 9999)
  16.  
  17.     /**
  18.       * 自己模拟一个黑名单（正常是用mysql，hbase，redis数据库读取出来的
  19.       */
  20.     //直接转化为RDD
  21.     val wordBlackList: RDD[(String, Boolean)] = ssc.sparkContext.parallelize(List("?", "!", "*"))
  22.       .map(param => (param, true))
  23.     /**
  24.       * (?,true)
  25.       * (!,true)
  26.       * (*,true)
  27.       */
  28.     val balckList: Array[(String, Boolean)] = wordBlackList.collect()
  29.     //broadcast广播出去
  30.     val blackListBroadcast: Broadcast[Array[(String, Boolean)]] = ssc.sparkContext.broadcast(balckList)
  31.  
  32.     /**
  33.       * 数据的处理
  34.       */
  35.     val wordOneDStream: DStream[(String, Int)] = dstream.flatMap(_.split(","))
  36.       .map((_, 1))
  37.     //transform把DStream转换成RDD,需要又返回值，并且类型为RDD
  38.     val wordCountDStream: DStream[(String, Int)] = wordOneDStream.transform(rdd => {
  39.       val filterRDD: RDD[(String, Boolean)] = rdd.sparkContext.parallelize(blackListBroadcast.value)
  40.       val resultRDD: RDD[(String, (Int, Option[Boolean]))] = rdd.leftOuterJoin(filterRDD)
  41.  
  42.       /**
  43.         * String, (Int, Option[Boolean])
  44.         * String：word
  45.         * Int：1
  46.         * Option：有可能join上也有可能join不上
  47.         *
  48.         * 思路：我们要的是join不上的，说白了要的是Option[Boolean]=None
  49.         * filter:
  50.         * true代表我们要
  51.         */
  52.       resultRDD.filter(tuple => {
  53.         tuple._2._2.isEmpty
  54.       }).map(_._1)
  55.     }).map((_, 1)).reduceByKey(_ + _)
  56.  
  57.     /**
  58.       * 数据的输出
  59.       */
  60.     wordCountDStream.print()
  61.  
  62.     ssc.start()
  63.     ssc.awaitTermination()
  64.     ssc.stop()
  65.   }
  66. }

先运行代码，然后再hadoop04传送数据




案例6：窗口操作Window Operations

  1. import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
  2. import org.apache.spark.{SparkConf, SparkContext}
  3. import org.apache.spark.streaming.{Seconds, StreamingContext}
  4.  
  5. /**
  6.   * 需求：每隔4秒统计最近6秒的单词计数的情况
  7.   * reduceByKeyAndWindow
  8.   */
  9. object WindowOperatorTest {
  10.   def main(args: Array[String]): Unit = {
  11.     val conf = new SparkConf().setMaster("local[2]").setAppName("WordBlack")
  12.     val sc: SparkContext = new SparkContext(conf)
  13.     val ssc = new StreamingContext(sc, Seconds(2))
  14.     /**
  15.       * 数据的输入
  16.       * 到目前为止这个地方还没有跟生产进行对接。
  17.       */
  18.     val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop04", 9999)
  19.  
  20.     /**
  21.       * 数据的处理
  22.       */
  23.     val resultWordCountDStream: DStream[(String, Int)] = dstream.flatMap(_.split(","))
  24.       .map((_, 1))
  25.  
  26.       /**
  27.         * reduceFunc: (V, V) => V,  匿名函数-达到对单词次数进行累加的效果
  28.         * windowDuration: Duration, 统计多少秒以内的数据-窗口的大小
  29.         * slideDuration: Duration,  每隔多少时间-滑动的大小
  30.         * //numPartitions: Int  指定分区数，要么跟核数有关要么和指定分区数有关
  31.         * 注意：这两个数一定要是Seconds(2)的倍数
  32.         */
  33.       .reduceByKeyAndWindow((x: Int, y: Int) => x + y, Seconds(6), Seconds(4))
  34.  
  35.  
  36.     /**
  37.       * 数据的输出
  38.       */
  39.     resultWordCountDStream.print()
  40.     ssc.start()
  41.     ssc.awaitTermination()
  42.     ssc.stop()
  43.   }
  44. }


foreachRDD的使用

注意foreachRDD下面的代码是执行在Driver端的

1.下图会报错

  1. dstream.foreachRDD { rdd =>
  2.   val connection = createNewConnection()  // executed at the driver
  3.   rdd.foreach { record =>
  4.     connection.send(record) // executed at the worker
  5.   }
  6. }

connection对象是需要发到executor上执行的。这就需要网络的传输了。就需要序列化。但是这个对象是不支持序列化的。所以就发送不过去，会报错无法序列化。


2.不报错，缺陷每次都需要创建connection对象，还有就是每次处理的是一条数据，频繁的创建和销毁对数据库连接，对数据库影响很大

  1. dstream.foreachRDD { rdd =>
  2.   rdd.foreach { record =>
  3.     val connection = createNewConnection()
  4.     connection.send(record)
  5.     connection.close()
  6.   }
  7. }

3.不报错，一次拿到一个partition数据，并创建一个connection对象。再把每个分区的数据一条一条发送到数据库。

  1. dstream.foreachRDD { rdd =>
  2.   rdd.foreachPartition { partitionOfRecords =>
  3.     val connection = createNewConnection()
  4.     partitionOfRecords.foreach(record => connection.send(record))
  5.     connection.close()
  6.   }
  7. }

调优：我们可以弄个连接池，再弄个批处理，每一百条数据提交一次到数据库

4.不报错，使用连接池

  1. dstream.foreachRDD { rdd =>
  2.   rdd.foreachPartition { partitionOfRecords =>
  3.     // ConnectionPool is a static, lazily initialized pool of connections
  4.     val connection = ConnectionPool.getConnection()
  5.     partitionOfRecords.foreach(record => connection.send(record))
  6.     ConnectionPool.returnConnection(connection)  // return to the pool for future reuse
  7.   }
  8. }

注意：从连接池中获取，使用完把连接丢回到连接池，少了创建的步骤，但是还是一条数据一条数据提交。所以我们要批处理，就要考虑到事务。

下面是操作实战：

现在mysql建立一张表：


写一个连接池的scala文件：

  1. import java.sql.{Connection, DriverManager}
  2.  
  3. object ConnectionPool {
  4.   private val max=8 ;//连接池的连接总数
  5.   private val connectionNum=10;//每次产生的连接数
  6.   private var conNum=0;//当前连接池已经产生的连接数
  7.  
  8.   import java.util
  9.   private val pool=new util.LinkedList[Connection]();//连接池
  10.  
  11.   {
  12.     Class.forName("com.mysql.jdbc.Driver")
  13.   }
  14.   /**
  15.     * 释放连接
  16.     */
  17.   def returnConnection(conn:Connection):Unit={
  18.     pool.push(conn);
  19.   }
  20.   /**
  21.     * 获取连接
  22.     */
  23.   def getConnection():Connection={
  24.     //同步代码块
  25.     AnyRef.synchronized({
  26.       if(pool.isEmpty()){
  27.         for( i <- 1 to connectionNum){
  28.           val conn=DriverManager.getConnection("jdbc:mysql://localhost:3306/aura","root","root");
  29.           pool.push(conn);
  30.           conNum+1;
  31.         }
  32.       }
  33.       pool.poll();
  34.     })
  35.  
  36.   }
  37. }

下面的代码：

  1. /**
  2.   * Create by jenrey on 2018/5/13 20:27
  3.   */
  4. import org.apache.spark.{SparkConf, SparkContext}
  5. import org.apache.spark.streaming.{Seconds, StreamingContext}
  6. import org.apache.spark.streaming.dstream.ReceiverInputDStream
  7.  
  8. /**
  9.   * 接收nc的数据，并把数据存到mysql表中
  10.   */
  11. object OutputTest {
  12.  
  13.   def main(args: Array[String]): Unit = {
  14.     val conf = new SparkConf().setMaster("local[2]").setAppName("OutputTest")
  15.     val sc = new SparkContext(conf)
  16.     val ssc = new StreamingContext(sc,Seconds(2))
  17.     ssc.checkpoint("hdfs://myha01/StreamingCheckPoint3")
  18.     /**
  19.       * 数据的输入
  20.       */
  21.     val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("hadoop04",9999)
  22.  
  23.  
  24.     val wordCountDStream = dstream.flatMap(_.split(","))
  25.       .map((_, 1))
  26.       .updateStateByKey((values: Seq[Int], state: Option[Int]) => {
  27.         val currentCount = values.sum
  28.         val lastCount = state.getOrElse(0)
  29.         Some(currentCount + lastCount)
  30.       })
  31.  
  32.     /**
  33.      * 数据的输出
  34.      */
  35.     wordCountDStream.foreachRDD( rdd=>{
  36.       rdd.foreachPartition( paritition =>{
  37.         //从连接池中获取连接
  38.         val connection = ConnectionPool.getConnection()
  39.         //获取Statement对象（用来发送sql指令）
  40.         val statement = connection.createStatement()
  41.         paritition.foreach{
  42.           case (word,count) =>{
  43.             val sql=s"insert into aura.1711wordcount values(now(),'$word',$count)"
  44.             print(sql)
  45.             //借助于Statement发送sql指令
  46.             statement.execute(sql)
  47.           }
  48.         }
  49.         //把connection对象再还回给连接池
  50.         ConnectionPool.returnConnection(connection)
  51.       } )
  52.     })
  53.  
  54.     ssc.start()
  55.     ssc.awaitTermination()
  56.     ssc.stop()
  57.   }
  58.  
  59. }
  60.

先运行程序再发送nc数据：



缺点就是把每一次的记录都打印出来了。

如果想要最新的数据就存在Hbase上面，Hbase会自动进行覆盖。


下面是将结果保存到Mysql的代码全集：
  1. import java.sql.DriverManager
  2.  
  3. import org.apache.spark.storage.StorageLevel
  4. import org.apache.spark.streaming.{Seconds, StreamingContext}
  5. import org.apache.spark.{SparkConf, SparkContext}
  6.  
  7. /**
  8.   * WordCount程序，Spark Streaming消费TCP Server发过来的实时数据的例子：
  9.   *
  10.   * 1、在master服务器上启动一个Netcat server
  11.   * `$ nc -lk 9998` (如果nc命令无效的话，我们可以用yum install -y nc来安装nc)
  12.   *
  13.   *
  14.   * create table wordcount(ts bigint, word varchar(50), count int);
  15.   *
  16.   * spark-shell --total-executor-cores 4 --executor-cores 2 --master spark://master:7077 --jars mysql-connector-java-5.1.44-bin.jar,c3p0-0.9.1.2.jar,spark-streaming-basic-1.0-SNAPSHOT.jar
  17.   *
  18.   *
  19.   */
  20. object NetworkWordCountForeachRDD {
  21.   def main(args: Array[String]) {
  22.     val sparkConf = new SparkConf().setAppName("NetworkWordCountForeachRDD")
  23.     val sc = new SparkContext(sparkConf)
  24.  
  25.     // Create the context with a 1 second batch size
  26.     val ssc = new StreamingContext(sc, Seconds(5))
  27.  
  28.     //创建一个接收器(ReceiverInputDStream)，这个接收器接收一台机器上的某个端口通过socket发送过来的数据并处理
  29.     val lines = ssc.socketTextStream("hadoop1", 9998, StorageLevel.MEMORY_AND_DISK_SER)
  30.  
  31.     //处理的逻辑，就是简单的进行word count
  32.     val words = lines.flatMap(_.split(" "))
  33.     val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)
  34.  
  35.     //将结果保存到Mysql(一)
  36.     /**
  37.       *
  38.       * 这个代码会报错的！！！
  39.       */
  40.     wordCounts.foreachRDD { (rdd, time) =>
  41.       Class.forName("com.mysql.jdbc.Driver")
  42.       val conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/aura", "root", "root")
  43.       val statement = conn.prepareStatement(s"insert into wordcount(ts, word, count) values (?, ?, ?)")
  44.       rdd.foreach { record =>
  45.         statement.setLong(1, time.milliseconds)
  46.         statement.setString(2, record._1)
  47.         statement.setInt(3, record._2)
  48.         statement.execute()
  49.       }
  50.       statement.close()
  51.       conn.close()
  52.     }
  53.     //启动Streaming处理流
  54.     ssc.start()
  55.  
  56.     ssc.stop(false)
  57.  
  58.  
  59.     //将结果保存到Mysql(二)
  60.     wordCounts.foreachRDD { (rdd, time) =>
  61.       rdd.foreach { record =>
  62.         Class.forName("com.mysql.jdbc.Driver")
  63.         val conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/aura", "root", "root")
  64.         val statement = conn.prepareStatement(s"insert into wordcount(ts, word, count) values (?, ?, ?)")
  65.         statement.setLong(1, time.milliseconds)
  66.         statement.setString(2, record._1)
  67.         statement.setInt(3, record._2)
  68.         statement.execute()
  69.         statement.close()
  70.         conn.close()
  71.       }
  72.     }
  73.  
  74.     //将结果保存到Mysql(三)
  75.     wordCounts.foreachRDD { (rdd, time) =>
  76.       rdd.foreachPartition { partitionRecords =>
  77.         Class.forName("com.mysql.jdbc.Driver")
  78.         val conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/aura", "root", "root")
  79.         val statement = conn.prepareStatement(s"insert into wordcount(ts, word, count) values (?, ?, ?)")
  80.         partitionRecords.foreach { case (word, count) =>
  81.           statement.setLong(1, time.milliseconds)
  82.           statement.setString(2, word)
  83.           statement.setInt(3, count)
  84.           statement.execute()
  85.         }
  86.         statement.close()
  87.         conn.close()
  88.       }
  89.     }
  90.  
  91.     //将结果保存到Mysql(四),使用连接池
  92.     wordCounts.foreachRDD { (rdd, time) =>
  93.       rdd.foreachPartition { partitionRecords =>
  94.         val conn = ConnectionPool.getConnection
  95.         val statement = conn.prepareStatement(s"insert into wordcount(ts, word, count) values (?, ?, ?)")
  96.         partitionRecords.foreach { case (word, count) =>
  97.           statement.setLong(1, time.milliseconds)
  98.           statement.setString(2, word)
  99.           statement.setInt(3, count)
  100.           statement.execute()
  101.         }
  102.         statement.close()
  103.         ConnectionPool.returnConnection(conn)
  104.       }
  105.     }
  106.  
  107.     //将结果保存到Mysql(五)，批处理
  108.     wordCounts.foreachRDD { (rdd, time) =>
  109.       rdd.foreachPartition { partitionRecords =>
  110.         val conn = ConnectionPool.getConnection
  111.         val statement = conn.prepareStatement(s"insert into wordcount(ts, word, count) values (?, ?, ?)")
  112.         partitionRecords.foreach { case (word, count) =>
  113.           statement.setLong(1, time.milliseconds)
  114.           statement.setString(2, word)
  115.           statement.setInt(3, count)
  116.           statement.addBatch()
  117.         }
  118.         statement.executeBatch()
  119.         statement.close()
  120.         ConnectionPool.returnConnection(conn)
  121.       }
  122.     }
  123.  
  124.  
  125.     //将结果保存到Mysql(六)，批处理引入事务
  126.     wordCounts.foreachRDD { (rdd, time) =>
  127.       rdd.foreachPartition { partitionRecords =>
  128.         val conn = ConnectionPool.getConnection
  129. 	//把自动提交改为false
  130.         conn.setAutoCommit(false)
  131.         val statement = conn.prepareStatement(s"insert into wordcount(ts, word, count) values (?, ?, ?)")
  132.         partitionRecords.foreach { case (word, count) =>
  133.           statement.setLong(1, time.milliseconds)
  134.           statement.setString(2, word)
  135.           statement.setInt(3, count)
  136.           statement.addBatch()
  137.         }
  138.         statement.executeBatch()
  139.         statement.close()
  140.         conn.commit()
  141.         conn.setAutoCommit(true)
  142.         ConnectionPool.returnConnection(conn)
  143.       }
  144.     }
  145.  
  146.  
  147.     //将结果保存到Mysql(七)，控制批处理的量，每500条提交一次
  148.     wordCounts.foreachRDD { (rdd, time) =>
  149.       rdd.foreachPartition { partitionRecords =>
  150.         val conn = ConnectionPool.getConnection
  151.         conn.setAutoCommit(false)
  152.         val statement = conn.prepareStatement(s"insert into wordcount(ts, word, count) values (?, ?, ?)")
  153.         partitionRecords.zipWithIndex.foreach { case ((word, count), index) =>
  154.           statement.setLong(1, time.milliseconds)
  155.           statement.setString(2, word)
  156.           statement.setInt(3, count)
  157.           statement.addBatch()
  158.           if (index != 0 && index % 500 == 0) {
  159.             statement.executeBatch()
  160.             conn.commit()
  161.           }
  162.         }
  163.         statement.executeBatch()
  164.         statement.close()
  165.         conn.commit()
  166.         conn.setAutoCommit(true)
  167.         ConnectionPool.returnConnection(conn)
  168.       }
  169.     }
  170.  
  171.     //等待Streaming程序终止
  172.     ssc.awaitTermination()
  173.   }
  174. }

案例7：SparkStreaming的数据源来自kafka

首先在pom.xml中写入下面的代码


编写下面的代码：

  1. import kafka.serializer.StringDecoder
  2. import org.apache.spark.streaming.dstream.DStream
  3. import org.apache.spark.streaming.kafka.KafkaUtils
  4. import org.apache.spark.{SparkConf, SparkContext}
  5. import org.apache.spark.streaming.{Seconds, StreamingContext}
  6.  
  7. object KafkaTest {
  8.  
  9.   def main(args: Array[String]): Unit = {
  10.     val conf = new SparkConf().setMaster("local[2]").setAppName("KafkaTest")
  11.     val sc = new SparkContext(conf)
  12.     val ssc = new StreamingContext(sc,Seconds(2))
  13.     //使用kafka是需要下面的目录的，因为SparkStreaming自己要维护一些东西的，要持久化，存到内存是易丢失的。
  14.     ssc.checkpoint("hdfs://myha01/streamingkafka")
  15.     /**
  16.       * 数据的输入：KafkaUtils.createDirectStream
  17.       *
  18.       * def createDirectStream[K: ClassTag,V: ClassTag,KD <:Decoder[K]: ClassTag,VD <:Decoder[V]: ClassTag] (
  19.       * 下面是三个参数：
  20.       * ssc: StreamingContext,
  21.       * kafkaParams: Map[String, String],
  22.       * topics: Set[String])  可以一下子读多个topics，但是我们这里读一个topics就行了
  23.       */
  24.       //指定kafka broker的机器，也就是kafka的地址
  25.     val kafkaParams = Map("metadata.broker.list" -> "hadoop03:9092")
  26.     val topics = Set("aura")
  27.     //kafka读出来数据是kv的形式[String代表k的数据类型（k可就是偏移位置的信息, String代表v的数据类型（kafka内每一条数据）, StringDecoder代表的就是解码器, StringDecoder]
  28.     //原来直接返回的是InputDStream[(String,String)]的KV数据类型，因为偏移位置的信息对我们是没有用的所以我们要.map(_._2)
  29.     val kafkaDStream: DStream[String] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
  30.       ssc, kafkaParams, topics).map(_._2)
  31.  
  32.     /**
  33.       * 数据的处理
  34.       * 也已经比较正式了
  35.       */
  36.     kafkaDStream.flatMap(_.split(","))
  37.       .map((_,1))
  38.       .reduceByKey(_+_)
  39.       .print()
  40.  
  41.     ssc.start()
  42.     ssc.awaitTermination()
  43.     ssc.stop()
  44.   }
  45. }


启动kafka高可用集群：

注意先启动ZK

[hadoop@hadoop03 kafka_2.11-0.8.2.0]$ bin/kafka-server-start.sh config/server.properties

[hadoop@hadoop04 kafka_2.11-0.8.2.0]$ bin/kafka-server-start.sh config/server.properties

[hadoop@hadoop05 kafka_2.11-0.8.2.0]$ bin/kafka-server-start.sh config/server.properties

创建topic

[hadoop@hadoop03 kafka_2.11-0.8.2.0]$ bin/kafka-topics.sh --create --zookeeper hadoop02:2181 --replication-factor 3 --partitions 3 --topic aura


创建生产者

[hadoop@hadoop03 kafka_2.11-0.8.2.0]$ bin/kafka-console-producer.sh --broker-list localhost:9092 -topic aura


然后运行代码：

然后再hadoop03输入下面的内容





案例8：数据黑名单过滤

  1. import java.sql.Date
  2. import java.util.Properties
  3.  
  4. import kafka.serializer.StringDecoder
  5. import org.apache.spark.rdd.RDD
  6. import org.apache.spark.sql.types.{LongType, StringType, StructField, StructType}
  7. import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}
  8. import org.apache.spark.streaming.dstream.DStream
  9. import org.apache.spark.streaming.kafka.KafkaUtils
  10. import org.apache.spark.streaming.{Seconds, StreamingContext}
  11. import org.apache.spark.{SparkConf, SparkContext}
  12.  
  13. /**
  14.   * timestamp:
  15.   * 时间戳，用户点击广告的时间
  16.   * province:
  17.   * 省份，用户在哪个省份点击的广告
  18.   * city:
  19.   * 城市，用户在哪个城市点击的广告
  20.   * userid:
  21.   * 用户的唯一标识
  22.   * advid:
  23.   * 被点击的广告id
  24.   */
  25.         object AdvApplicationTest {
  26.  
  27.           def main(args: Array[String]): Unit = {
  28.             val conf = new SparkConf()
  29.             conf.setMaster("local")
  30.             conf.setAppName("AdvApplicationTest")
  31.             conf.set("","")  //序列化
  32.  
  33.             val sc = new SparkContext(conf)
  34.  
  35.             val ssc = new StreamingContext(sc,Seconds(5))
  36.             //getOrCreate()：有就拿过来，没有就创建，类似于单例模式：
  37.             val spark: SparkSession = SparkSession.builder()
  38.               .config(conf).getOrCreate()
  39.  
  40.             /**
  41.               * 第一步：从kafka获取数据（direct  方式）
  42.               *   K: ClassTag,
  43.                   V: ClassTag,
  44.                   KD <: Decoder[K]: ClassTag,
  45.                   VD <: Decoder[V]: ClassTag] (
  46.               ssc: StreamingContext,
  47.               kafkaParams: Map[String, String],
  48.               topics: Set[String]
  49.               */
  50.             val kafkaParams = Map("metadata.broker.list" -> "hadoop3:9092")
  51.             val topics = Set("aura")
  52.             val logDstream: DStream[String] = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
  53.               ssc, kafkaParams, topics).map(_._2)
  54.  
  55.             /**
  56.               * 第二步：进行黑名单过滤
  57.               */
  58.             val filterLogDStream: DStream[String] = blackListFilter(logDstream,ssc)
  59.  
  60.  
  61.             /**
  62.               * 【一个用户】【一天内】对【某个广告】点击的次数超过了【100次】，这样的用户属于黑名单用户
  63.               *
  64.               *
  65.               * zhangsan:
  66.               *          A:50  B:60
  67.               * lisi:
  68.               *          A:50   A:20  A:40   这就是黑名单用户
  69.               * 如果一个用户今天是黑名单用户，那么明天还是黑名单用户吗？
  70.               * 这个看业务而定。
  71.               *
  72.               * 第三步：动态生成黑名单  实时生成黑名单
  73.               */
  74.             DynamicGenerationBlacklists(filterLogDStream,spark)
  75.  
  76.             /**
  77.               * 第四步：
  78.               *        实时统计每天各省各城市广告点击量
  79.               */
  80.             val dateProvinceCityAdvClick_Count = ProvinceCityAdvClick_Count(filterLogDStream)
  81.             /**
  82.               * 第五步：
  83.               *       实时统计每天各省热门广告
  84.               *        分组求TopN
  85.               *
  86.               *   transform  froeachRDD
  87.               *   rdd   => dataframe
  88.               *   SparkSQL:
  89.               *     SQL
  90.               */
  91.  
  92.  
  93.             /**
  94.               * 第六步：
  95.               *     实时统计每天每个广告在最近一小时的滑动窗口的点击趋势
  96.               */
  97.  
  98.             ssc.start()
  99.             ssc.awaitTermination()
  100.             ssc.stop()
  101.           }
  102.  
  103.           /**
  104.             * 对黑名单数据进行过滤
  105.             * logDstream  从kafka读取数据
  106.             * 返回的就是进行黑名单过滤以后的数据
  107.             */
  108.           def blackListFilter(logDstream: DStream[String],ssc:StreamingContext):DStream[String]={
  109.  
  110.             /**
  111.               * 这个地方应该是去数据库里面去读取数据
  112.               * 三个常用的数据库：Redis，HBase，Mysql
  113.               * black_list
  114.               */
  115.  
  116.             val blackList = List((1L,true),(2L,true),(3L,true))
  117.             val blackListRDD = ssc.sparkContext.parallelize(blackList)
  118.             val balckListBroadcast = ssc.sparkContext.broadcast(blackListRDD.collect())
  119.  
  120.             /**
  121.               * 这个地方的黑名单，应该是从我们的持久化的数据库里面读取的：有三个数据库是我们常用的：
  122.               * 1）Reids   自己去百度一下
  123.               * 2) HBase  自己去百度一下
  124.               * 3) Mysql  上课演示过
  125.               * SparkCore的方式读取的
  126.               * SparkSQL  -> dataframe -> rdd
  127.               */
  128.  
  129.             logDstream.transform( rdd =>{
  130.              val user_lineRDD=rdd.map( line =>{
  131.                val fields = line.split(",")
  132.                (fields(3).toLong,line)
  133.              })
  134.                val blackRDD = rdd.sparkContext.parallelize(balckListBroadcast.value)
  135.               //只有keyValue的形式才能进行join，所以需要上面的操作
  136.               val resultRDD: RDD[(Long, (String, Option[Boolean]))] = user_lineRDD.leftOuterJoin(blackRDD)
  137.               resultRDD.filter( tuple =>{
  138.                 tuple._2._2.isEmpty
  139.               }).map(_._2._1)
  140.  
  141.             })
  142.  
  143.           }
  144.  
  145.           /**
  146.             * 动然生成黑名单
  147.             * @param filterLogDStream  黑名单过滤万了以后的数据
  148.             * 【一个用户】【一天内】对【某个广告】点击的次数超过了【100次】，这样的用户属于黑名单用户
  149.             *
  150.             * 梳理一下思路：
  151.             *   这个需求 跟 我们单词计数很像，无非不就是实时统计每个单词出现了多少次
  152.             *   如果发现某个单词出现了一个100，那么他就是黑名单单词
  153.             *   方式一：
  154.             *   (date_userid_advid,v)=map
  155.             *    实时统计出来每个单词出现了多少次=updateStateBykey （对内存的要求高一点）
  156.             *    张三 A 80
  157.             *    李四 B 99
  158.             *         100
  159.             *    fitler  过滤出来次数 一百以上 把它写入 MySQL，Reids,HBase 数据库
  160.             *   方式二：
  161.             *   (date_userid_advid,v)=map
  162.             *    每次处理的是本批次的数据 reduceBykey(对内存的要求低一点)
  163.             *    HBase:
  164.             *        rowkey:  date_userid_advid  2
  165.             *          本批次  3
  166.             *            5
  167.             *    Redis
  168.             *   方式三：
  169.             *        MySQL的方式
  170.             */
  171.           def DynamicGenerationBlacklists(filterLogDStream: DStream[String],spark:SparkSession):Unit={
  172.  
  173.             val date_userid_advid_ds=filterLogDStream.map( line =>{
  174.               val fields = line.split(",")
  175.              val time = new Date( fields(0).toLong)
  176.               val date = DateUtils.formatDateKey(time)
  177.               val userid = fields(3)
  178.               val advid = fields(4)
  179.                //20180512_
  180.               (date+"_"+userid+"_"+advid,1L)
  181.             }).reduceByKey(_+_)
  182.  
  183.             date_userid_advid_ds.foreachRDD( rdd =>{
  184.               rdd.foreachPartition( partition =>{
  185.                 val connection = ConnectionPool.getConnection()
  186.                 val statement = connection.createStatement()
  187.                 partition.foreach{
  188.                   case(date_userid_advid,count) =>{
  189.                     val fields = date_userid_advid.split("_")
  190.                     val date = fields(0)
  191.                     val userid = fields(1).toLong
  192.                     val advid = fields(2).toLong
  193.                     val sql=s"insert into aura.tmp_advclick_count values($date,$userid,$advid,$count)";
  194.                     statement.execute(sql);
  195.                   }
  196.                 }
  197.                 ConnectionPool.returnConnection(connection)
  198.  
  199.               })
  200.             })
  201.  
  202.             /**
  203.               *生成黑名单
  204.               */
  205.  
  206.             val df: DataFrame = spark.read.format("jdbc")
  207.               .option("url", "jdbc:mysql://localhost:3306/aura")
  208.               .option("user", "aura")
  209.               .option("password", "aura")
  210.               .option("dbtable", "tmp_advclick_count")
  211.               .load()
  212.  
  213.             df.createOrReplaceTempView("tmp_advclick_count")
  214.  
  215.             val sql=
  216.               """
  217.                  SELECT
  218.                       userid
  219.                  FROM
  220.                  (
  221.                  SELECT
  222.                       date,userid,advid,sum(click_count) c_count
  223.                       FROM
  224.                       tmp_advclick_count
  225.                  GROUP BY
  226.                       date,userid,advid
  227.                  ) t
  228.                       WHERE
  229.                       t.c_count > 100
  230.               """
  231.  
  232.             //统计出来黑名单
  233.             val blacklistdf = spark.sql(sql).distinct()
  234.               val properties = new Properties()
  235.             properties.put("user","aura")
  236.             properties.put("password","aura")
  237.             blacklistdf.write.mode(SaveMode.Append)
  238.                 .jdbc("jdbc:mysql://localhost:3306/aura","black_list",properties)
  239.           }
  240.  
  241.           /**
  242.             * 实时统计每天各省各城市广告点击量
  243.             * @param filterLogDStream
  244.             */
  245.           def ProvinceCityAdvClick_Count(filterLogDStream: DStream[String]):DStream[(String,Long)]={
  246.             /**
  247.               * 思路
  248.               * map  => (k,v)  => date+province+city+advid  1
  249.               *                updateStateBykey
  250.               */
  251.             var f=(input:Seq[Long],state:Option[Long]) =>{
  252.               val current_count = input.sum
  253.               val last_count = state.getOrElse(0)
  254.               Some(current_count+last_count)
  255.             }
  256.  
  257.             filterLogDStream.map( line =>{
  258.               val fields = line.split(",")
  259.               val time = fields(0).toLong
  260.               val mydate = new Date(time)
  261.               val date = DateUtils.formatDateKey(mydate)
  262.               val province = fields(1)
  263.               val city = fields(2)
  264.               val advid = fields(4)
  265.               (date+"_"+province+"_"+city+"_"+advid,1L)
  266.             }).updateStateByKey(f)
  267.             /**
  268.               * 如果开发有需求的话，可以把这些数据库写入 MySQL数据库 ，Hbase
  269.               */
  270.           }
  271.  
  272.           /**
  273.             * 实时统计 各省热门广告
  274.             *
  275.             * transform : rdd  -> datafram  -> table -> sql
  276.             * @param date_province_city_advid_count
  277.             */
  278.           def ProvinceAdvClick_Count(date_province_city_advid_count:DStream[(String,Long)],spark:SparkSession): Unit ={
  279.             date_province_city_advid_count.transform( rdd =>{
  280.             var date_province_advid_count=  rdd.map{
  281.                 case(date_province_city_advid,count) =>{
  282.                   val fields = date_province_city_advid.split("_")
  283.                   val date = fields(0)
  284.                   val province = fields(1)
  285.                   val advid = fields(3)
  286.  
  287.  
  288.                   (date+"_"+province+"_"+advid,count)
  289.                 }
  290.               }.reduceByKey(_+_)
  291.  
  292.              val rowRDD=date_province_advid_count.map( tuple =>{
  293.                 val fields = tuple._1.split("_")
  294.                 val date = fields(0)
  295.                 val provnice = fields(1)
  296.                 val advid = fields(2).toLong
  297.                 val count = tuple._2
  298.                 Row(date,provnice,advid,count)
  299.               })
  300.  
  301.               val schema=StructType(
  302.                 StructField("date",StringType,true)::
  303.                   StructField("province",StringType,true)::
  304.                   StructField("advid",LongType,true)::
  305.                   StructField("count",LongType,true):: Nil
  306.  
  307.               )
  308.  
  309.               val df = spark.createDataFrame(rowRDD,schema)
  310.  
  311.               df.createOrReplaceTempView("temp_date_province_adv_count")
  312.  
  313.               val sql=
  314.                 """
  315.                    select
  316.                         *
  317.                    from
  318.                    (
  319.                    select
  320.                         date,province,advid,count,row_number() over(partition by province ordr by count desc) rank
  321.                    from
  322.                         temp_date_province_adv_count
  323.                    ) temp
  324.                    where temp.rank < 10
  326.                 """
  327.  
  328.               /**
  329.                 * 把结果持久化到数据库
  330.                 */
  331.               spark.sql(sql)
  332.  
  333.              rdd
  334.  
  335.             })
  336.           }
  337.         }
  1. import java.text.ParseException;
  2. import java.text.SimpleDateFormat;
  3. import java.util.Calendar;
  4. import java.util.Date;
  5.  
  6. /**
  7.  * DateTime Utils
  8.  *
  9.  * Created by XuanYu on 2016/5/31.
  10.  */
  11. public class DateUtils {
  12.  
  13.     public static final SimpleDateFormat TIME_FORMAT =
  14.             new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  15.     public static final SimpleDateFormat DATE_FORMAT =
  16.             new SimpleDateFormat("yyyy-MM-dd");
  17.     public static final SimpleDateFormat DATEKEY_FORMAT =
  18.             new SimpleDateFormat("yyyyMMdd");
  19.  
  20.     /**
  21.      * 判断一个时间是否在另一个时间之前
  22.      * @param time1 第一个时间
  23.      * @param time2 第二个时间
  24.      * @return 判断结果
  25.      */
  26.     public static boolean before(String time1, String time2) {
  27.         try {
  28.             Date dateTime1 = TIME_FORMAT.parse(time1);
  29.             Date dateTime2 = TIME_FORMAT.parse(time2);
  30.  
  31.             if(dateTime1.before(dateTime2)) {
  32.                 return true;
  33.             }
  34.         } catch (Exception e) {
  35.             e.printStackTrace();
  36.         }
  37.         return false;
  38.     }
  39.  
  40.     /**
  41.      * 判断一个时间是否在另一个时间之后
  42.      * @param time1 第一个时间
  43.      * @param time2 第二个时间
  44.      * @return 判断结果
  45.      */
  46.     public static boolean after(String time1, String time2) {
  47.         try {
  48.             Date dateTime1 = TIME_FORMAT.parse(time1);
  49.             Date dateTime2 = TIME_FORMAT.parse(time2);
  50.  
  51.             if(dateTime1.after(dateTime2)) {
  52.                 return true;
  53.             }
  54.         } catch (Exception e) {
  55.             e.printStackTrace();
  56.         }
  57.         return false;
  58.     }
  59.  
  60.     /**
  61.      * 计算时间差值（单位为秒）
  62.      * @param time1 时间1
  63.      * @param time2 时间2
  64.      * @return 差值
  65.      */
  66.     public static int minus(String time1, String time2) {
  67.         try {
  68.             Date datetime1 = TIME_FORMAT.parse(time1);
  69.             Date datetime2 = TIME_FORMAT.parse(time2);
  70.  
  71.             long millisecond = datetime1.getTime() - datetime2.getTime();
  72.  
  73.             return Integer.valueOf(String.valueOf(millisecond / 1000));
  74.         } catch (Exception e) {
  75.             e.printStackTrace();
  76.         }
  77.         return 0;
  78.     }
  79.  
  80.     /**
  81.      * 获取年月日和小时
  82.      * @param datetime 时间（yyyy-MM-dd HH:mm:ss）
  83.      * @return 结果（yyyy-MM-dd_HH）
  84.      */
  85.     public static String getDateHour(String datetime) {
  86.         String date = datetime.split(" ")[0];
  87.         String hourMinuteSecond = datetime.split(" ")[1];
  88.         String hour = hourMinuteSecond.split(":")[0];
  89.         return date + "_" + hour;
  90.     }
  91.  
  92.     /**
  93.      * 获取当天日期（yyyy-MM-dd）
  94.      * @return 当天日期
  95.      */
  96.     public static String getTodayDate() {
  97.         return DATE_FORMAT.format(new Date());
  98.     }
  99.  
  100.     /**
  101.      * 获取昨天的日期（yyyy-MM-dd）
  102.      * @return 昨天的日期
  103.      */
  104.     public static String getYesterdayDate() {
  105.         Calendar cal = Calendar.getInstance();
  106.         cal.setTime(new Date());
  107.         cal.add(Calendar.DAY_OF_YEAR, -1);
  108.  
  109.         Date date = cal.getTime();
  110.  
  111.         return DATE_FORMAT.format(date);
  112.     }
  113.  
  114.     /**
  115.      * 格式化日期（yyyy-MM-dd）
  116.      * @param date Date对象
  117.      * @return 格式化后的日期
  118.      */
  119.     public static String formatDate(Date date) {
  120.         return DATE_FORMAT.format(date);
  121.     }
  122.  
  123.     /**
  124.      * 格式化时间（yyyy-MM-dd HH:mm:ss）
  125.      * @param date Date对象
  126.      * @return 格式化后的时间
  127.      */
  128.     public static String formatTime(Date date) {
  129.         return TIME_FORMAT.format(date);
  130.     }
  131.  
  132.     /**
  133.      * 解析时间字符串
  134.      * @param time 时间字符串
  135.      * @return Date
  136.      */
  137.     public static Date parseTime(String time) {
  138.         try {
  139.             return TIME_FORMAT.parse(time);
  140.         } catch (ParseException e) {
  141.             e.printStackTrace();
  142.         }
  143.         return null;
  144.     }
  145.  
  146.     /**
  147.      * 格式化日期key
  148.      * @param date
  149.      * @return
  150.      */
  151.     public static String formatDateKey(Date date) {
  152.         return DATEKEY_FORMAT.format(date);
  153.     }
  154.  
  155.     /**
  156.      * 格式化日期key
  157.      * @param datekey
  158.      * @return
  159.      */
  160.     public static Date parseDateKey(String datekey) {
  161.         try {
  162.             return DATEKEY_FORMAT.parse(datekey);
  163.         } catch (ParseException e) {
  164.             e.printStackTrace();
  165.         }
  166.         return null;
  167.     }
  168.  
  169.     /**
  170.      * 格式化时间，保留到分钟级别
  171.      * yyyyMMddHHmm
  172.      * @param date
  173.      * @return
  174.      */
  175.     public static String formatTimeMinute(Date date) {
  176.         SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMddHHmm");
  177.         return sdf.format(date);
  178.     }
  179.  
  180. }
  1. import java.sql.{Connection, DriverManager}
  2. object ConnectionPool {
  3.   private val max=8 ;//连接池的连接总数
  4.   private val connectionNum=10;//每次产生的连接数
  5.   private var conNum=0;//当前连接池已经产生的连接数
  6.  
  7.   import java.util
  8.   private val pool=new util.LinkedList[Connection]();//连接池
  9.  
  10.   {
  11.  
  12.     Class.forName("com.mysql.jdbc.Driver")
  13.   }
  14.   /**
  15.    * 释放连接
  16.    */
  17.   def returnConnection(conn:Connection):Unit={
  18.     pool.push(conn);
  19.   }
  20.   /**
  21.    * 获取连接
  22.    */
  23.   def getConnection():Connection={
  24.     //同步代码块
  25.     AnyRef.synchronized({
  26.       if(pool.isEmpty()){
  27.         for( i <- 1 to connectionNum){
  28.           val conn=DriverManager.getConnection("jdbc:mysql://localhost:3306/aura","root","root");
  29.           pool.push(conn);
  30.           conNum+1;
  31.         }
  32.       }
  33.  
  34.       pool.poll();
  35.     })
  36.  
  37.   }
  38. }







声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/正经夜光杯/article/detail/945515
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号