1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
package example8

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession

object StructuredStreamingQuickStart {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[4]").setAppName("StructuredNetworkWordCount")
    val sc = new SparkContext(conf)
    sc.setLogLevel("FATAL")
    val spark = SparkSession.builder().getOrCreate()
    import spark.implicits._
    // 构建DataFrame
    val df = spark.readStream.format("socket").option("host","gaozhy").option("port","9999").load()
		
    // 转换为Dataset 进行算子操作
    val words = df.as[String].flatMap(_.split(" "))

    val wordCounts = words.groupBy("value").count()

    val query = wordCounts.writeStream.outputMode("complete").format("console").start()

    query.awaitTermination()
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
-------------------------------------------
Batch: 0
-------------------------------------------
+-----+-----+
|value|count|
+-----+-----+
+-----+-----+

-------------------------------------------
Batch: 1
-------------------------------------------
+-----+-----+
|value|count|
+-----+-----+
|   AA|    3|
|   CC|    1|
|   BB|    2|
+-----+-----+

-------------------------------------------
Batch: 2
-------------------------------------------
+-----+-----+
|value|count|
+-----+-----+
|   AA|    5|
|   CC|    1|
|   BB|    2|
+-----+-----+

...

1
2
3
4
5
<dependency>
  <groupId>org.apache.spark</groupId>
  <artifactId>spark-sql-kafka-0-10_2.11</artifactId>
<version>2.4.4</version>

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
package example8

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.StructType

object StructuredStreamingSource {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[4]").setAppName("StructuredNetworkWordCount")
    val sc = new SparkContext(conf)
    sc.setLogLevel("FATAL")
    val spark = SparkSession.builder().getOrCreate()
    import spark.implicits._
    // =====================================CSV===========================================
    /*
    val df = spark.readStream
      .format("csv")
      //.option("sep",";")
      .schema(new StructType().add("id", "integer").add("name", "string").add("salary", "double"))
      .csv("/Users/gaozhy/data/csv")
     */
    // =====================================CSV===========================================

    // =====================================json===========================================
    /*
    val df = spark.readStream
      .format("json")
      .schema(new StructType().add("id","integer").add("name","string").add("salary","float"))
      .json("/Users/gaozhy/data/json")
    */
    // =====================================json===========================================

    //df.createOrReplaceTempView("t_user")
    /*
    spark.sql("select * from t_user")
      .writeStream
      .outputMode("append")
      .format("console")
      .start()
      .awaitTermination()
    */
    
    // =====================================kafka===========================================
    val df = spark
      .readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "gaozhy:9092")
      .option("startingOffsets", """{"baizhi":{"0":-2}}""") // 指定偏移量消费
      .option("subscribe", "baizhi")
      .load()

    val kafka = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "topic", "partition")
      .as[(String, String, String, Int)]

    kafka.createOrReplaceTempView("t_kafka")
//    spark.sql("select * from t_kafka")
//      .writeStream
//      .format("console")
//      .outputMode("append")
//      .start()
//      .awaitTermination()

    spark.sql("select count(*) from t_kafka group by partition")
      .writeStream
      .format("console")
      .outputMode("complete")
      .start()
      .awaitTermination()
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
package com.baizhi

import org.apache.spark.sql.{ForeachWriter, Row, SparkSession}
import org.apache.spark.sql.streaming.OutputMode
import redis.clients.jedis.Jedis

/**
  * 如何通过不同的sink构建结构化流计算结构的输出
  */
object SparkStructuredStreamingForOutputSink {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("input source")
      .getOrCreate()

    spark.sparkContext.setLogLevel("ERROR")
    import spark.implicits._


    //--------------------------------kafka【重点】-----------------------------------
    val df = spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "HadoopNode01:9092,HadoopNode02:9092,HadoopNode03:9092")
      .option("startingOffsets", """{"baizhi":{"0":-2}}""") // 指定消费baizhi topic的0号分区的earliest消费方式
      // 指定偏移量消费  In the json, -2 as an offset can be used to refer to earliest, -1 to latest.
      .option("subscribe", "baizhi")
      .load

    // kafka record 转换为要求的类型
    //    val ds = df
    //      .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "CAST(topic AS STRING)", "CAST(partition AS INT)", "CAST(timestamp AS LONG)")
    //      .as[(String, String, String, Int, Long)] // key value topic partition Long
    //
    //    ds.createOrReplaceTempView("t_kafka")
    //
    //    val text = spark.sql("select key as k,value as v,topic as t,partition as p, timestamp as ts from t_kafka")
    //--------------------------------kafka【重点】-----------------------------------

    //    text
    //      .writeStream
    //      .format("console")
    //      .outputMode(OutputMode.Append())
    //      .start()
    //      .awaitTermination()

    //===================================文件【输出模式：只支持Append】===========================================
    //    text
    //      .writeStream
    //      .format("json") // 文件格式 CSV JSON parquet ORC等
    //      .outputMode(OutputMode.Append())
    //      .option("checkpointLocation","hdfs://Spark:9000/checkpoint1")     // 检查点path 用以故障恢复
    //      .option("path","file:///D://result") // path支持本地和HDFS文件系统路径
    //      .start()
    //      .awaitTermination()

    //    text
    //      .writeStream
    //      .format("csv") // 文件格式 CSV JSON parquet ORC等
    //      .outputMode(OutputMode.Append())
    //      .option("checkpointLocation", "hdfs://Spark:9000/checkpoint2") // 检查点path 用以故障恢复
    //      .option("path", "file:///D://result2") // path支持本地和HDFS文件系统路径
    //      .start()
    //      .awaitTermination()


    //===================================Kafka【输出模式：Append | Updated | Completed】=========================
    //    val ds = df
    //      .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "CAST(topic AS STRING)", "CAST(partition AS INT)", "CAST(timestamp AS LONG)")
    //      .as[(String, String, String, Int, Long)].flatMap(_._2.split(" ")).map((_, 1)) // key value topic partition Long
    //
    //    ds.createOrReplaceTempView("t_kafka")
    //
    //    val text = spark.sql("select _1 as word,count(_2) as num from t_kafka group by _1")
    //
    //    text
    //      // .selectExpr("CAST(k AS STRING) as key", "CAST(v AS STRING) as value") // 对输出到kafka的数据定义key 和 value信息
    //      .selectExpr("CAST(word AS STRING) as key", "CAST(num AS STRING) as value") // 对输出到kafka的数据定义key 和 value信息
    //      .writeStream
    //      .format("kafka") // 文件格式 CSV JSON parquet ORC等
    //      // .outputMode(OutputMode.Append())
    //      .outputMode(OutputMode.Update())
    //      .option("checkpointLocation", "hdfs://Spark:9000/checkpoint4") // 检查点path 用以故障恢复
    //      .option("kafka.bootstrap.servers", "HadoopNode01:9092,HadoopNode02:9092,HadoopNode03:9092") // kafka集群信息
    //      .option("topic", "result") // 指定计算结果的保存topic
    //      .start()
    //      .awaitTermination()

    /*
      [root@HadoopNode02 kafka_2.11-2.2.0]# bin/kafka-console-consumer.sh --topic result --bootstrap-server HadoopNode01:9092,HadoopNode02:9092,HadoopNode03:9092 --property print.key=true
      */

    //===================================Foreach【输出模式：Append | Updated | Completed】==================================
    // 将计算结果输出到redis中
    //kafka record 转换为要求的类型

    val ds = df
      .selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)", "CAST(topic AS STRING)", "CAST(partition AS INT)", "CAST(timestamp AS LONG)")
      .as[(String, String, String, Int, Long)].flatMap(_._2.split(" ")).map((_, 1)) // key value topic partition Long

    ds.createOrReplaceTempView("t_kafka")

    val text = spark.sql("select _1 as word,count(_2) as num from t_kafka group by _1")
    text
      .writeStream
      .outputMode(OutputMode.Update())
      .foreach(new ForeachWriter[Row] {
        /**
          * 打开方法
          *
          * @param partitionId 分区序号
          * @param epochId
          * @return Boolean  true 创建一个连接 对这一行数据进行处理操作
          *         false 不会创建连接 跳过这一行数据
          */
        override def open(partitionId: Long, epochId: Long): Boolean = true

        /**
          * 处理方法
          *
          * @param value resultTable 行对象
          */
        override def process(value: Row): Unit = {
          val word = value.getString(0)
          val count = value.getLong(1).toString

          val jedis = new Jedis("Spark", 6379)
          jedis.set(word, count)
          jedis.close()
        }

        override def close(errorOrNull: Throwable): Unit = if (errorOrNull != null) errorOrNull.printStackTrace()

      }) // 对resultTable中的每一行记录应用写出规则
      .option("checkpointLocation", "hdfs://Spark:9000/checkpoint4") // 检查点path 用以故障恢复
      .start()
      .awaitTermination()
  }
}

1
2

1
2
3
4
5
6
7
8
9
10
11
12
import spark.implicits._

val words = ... // streaming DataFrame of schema { timestamp: Timestamp, word: String }

// Group the data by window and word and compute the count of each group
val windowedCounts = words
    .withWatermark("timestamp", "10 minutes")
    .groupBy(
        window($"timestamp", "10 minutes", "5 minutes"),
        $"word")
    .count()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
package com.baizhi.window

import java.sql.Timestamp
import java.text.SimpleDateFormat

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.streaming.OutputMode

/**
  * 基于事件事件窗口操作
  *
  */
object SparkStructuredStreamingForLateDataHandling {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("window on event time").master("local[*]").getOrCreate()
    spark.sparkContext.setLogLevel("ERROR")
    import spark.implicits._

    val df = spark
      .readStream
      .format("socket")
      .option("host", "Spark")
      .option("port", "4444")
      .load()

    // 发送数据时：word,timestamp
    // Hello,1573525440000  --> 10:24:00

    // 导入隐式转换的函数
    import org.apache.spark.sql.functions._

    val sdf = new SimpleDateFormat("HH:mm:ss") // 时:分:秒

    df
      .as[String]
      .map(_.split(",")) // Hello,1573525440000 --> Array("Hello","1573525440000")
      .map(arr => (arr(0), new Timestamp(arr(1).toLong))) //("Hello",new Timestamp(1573525440000))
      .toDF("word", "timestamp")
      //-------------------------------------------------------------
      // 水位线操作
      .withWatermark("timestamp", "10 seconds") // vm = maxEventTime-2
      //-------------------------------------------------------------

      // 先根据事件时间划分窗口，再对窗口内的单词进一步分组
      .groupBy(window($"timestamp", "10 seconds", "5 seconds"), $"word") // 滑动窗口 长度5s  滑动3s
      // 统计窗口内单词出现的次数
      .count()

      /**
        * root
        * |-- window: struct (nullable = true)
        * |    |-- start: timestamp (nullable = true)
        * |    |-- end: timestamp (nullable = true)
        * |-- word: string (nullable = true)
        * |-- count: long (nullable = false)
        */
      //.printSchema()
      .map(row => (sdf.format(row.getStruct(0).getTimestamp(0)), sdf.format(row.getStruct(0).getTimestamp(1)), row.getString(1), row.getLong(2)))
      .toDF("startTime", "endTime", "word", "num")

      .writeStream
      .format("console")
      // .outputMode(OutputMode.Append()) // 注意：输出模式只能是 append和 update
      .outputMode(OutputMode.Update()) // 注意：输出模式只能是 append和 update
      .start()
      .awaitTermination()
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.types.{BooleanType, StructType}

object SparkStructuredStreamingForJoinOpt {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("join opt").master("local[*]").getOrCreate()

    val df1 = spark   // 基于静态文件创建DF
      .read
      .format("json")
      .load("file:///G:\IDEA_WorkSpace\scala-workspace\spark-day11\src\main\resources") // id name sex

    val df2 = spark  // 基于流数据创建DF
      .readStream
      .format("csv")
      .schema(
        new StructType()
          .add("id", "integer")
          .add("name", "string")
          .add("sex", BooleanType)
          .add("salary", "double")
      )
      .csv("file:///d://csv")

    // 批和流不允许Join【流数据不能join给批数据】
    // 正常：         【批数据join给流数据】
    df2.join(df1,Seq("id","id"),"leftOuter")  // 流DF join 批DF
      .writeStream
      .format("console")
      .outputMode("append")
      .start()
      .awaitTermination()
  }
}

//----------------------------------------------------------------------
Batch: 0
-------------------------------------------
+---+---+----+------+----+------+
| id| id|name|salary|name|salary|
+---+---+----+------+----+------+
|  1|  1|  zs|1000.0|  zs|3000.0|
|  2|  2|  ls|2000.0|  ls|3000.0|
|  3|  3|  ww|3000.0|  ww|3000.0|
|  4|  4|  zs|1000.0| zs2|3000.0|
|  5|  5|  ls|2000.0| ls2|3000.0|
|  6|  6|  ww|3000.0| ww2|3000.0|
|  4|  4|  zs|3000.0| zs2|3000.0|
|  7|  7|  ls|2000.0|null|  null|
|  8|  8|  ww|3000.0|null|  null|
+---+---+----+------+----+------+