Flink(scala)整合MySQL統計UV(unique visitor)

資料源是尚硅谷的課件, 需要的話可以私信我

核心代碼

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector

import java.sql.{Connection, DriverManager, PreparedStatement, Timestamp}
import java.text.SimpleDateFormat
import java.util.Properties



// 每條資料
/*
83.149.9.216 - - 17/05/2015:10:05:03 +0000 GET /presentations/logstash-monitorama-2013/images/kibana-search.png
83.149.9.216 - - 17/05/2015:10:05:43 +0000 GET /presentations/logstash-monitorama-2013/images/kibana-dashboard3.png
83.149.9.216 - - 17/05/2015:10:05:47 +0000 GET /presentations/logstash-monitorama-2013/plugin/highlight/highlight.js
*/
// 輸入樣例類
case class UVItem(url: String, ip:String, timestamp: Long)
// 基于WindowEnd分組的樣例類
case class UVWindowEnd(url: String, WindowEnd: Long, Count: Long)
// 目標 每五分鐘統計這個1小時的每個頁面的UV值
object UniqueVisitor {
  def main(args: Array[String]): Unit = {
    // 創建環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 設定時間特性為事件時間
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // kafka消費資料
    /*
    // 配置kafka
    val properties = new Properties()
    properties.put("bootstrap.server", "kafka的ip地址")
    // 從kafka消費資料
    val inputStream = env.addSource(new FlinkKafkaConsumer[String]("訂閱主題",new SimpleStringSchema() ,properties))
    */
    // 讀取resource的資料檔案
    val inputStream = env.readTextFile(getClass.getResource("/apache.log").getPath)
    // 將每行資料用空格切割后 封裝成樣例類 資料亂序 并指定時間戳 設定Watermark為 30秒
    val dataStream = inputStream
      .map(data=>{
        val arr = data.split(" ")
        val timestamp = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss").parse(arr(3)).getTime
        // (url: String, ip:String, timestamp: Long)
        UVItem(arr(6), arr(0), timestamp)
      }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[UVItem](Time.seconds(30)) {
      override def extractTimestamp(t: UVItem): Long = t.timestamp
    })

    dataStream
      .keyBy(_.url) // url作為key進行分組
      .timeWindow(Time.hours(1), Time.minutes(5)) // 開滾動視窗 長度1小時 步長5分鐘
      .process(new CountUVProcess()) // 自定義類繼承ProcessWindowFunction 對每個url進行統計 (url: String, WindowEnd: Long, Count: Long)
      .keyBy(_.WindowEnd) // 視窗結束時間作為key進行分組
      .process(new windowEndProcess()) // 對每個視窗的資料包裝成要存到MySQL的元組 (Long, String, Long)(視窗結束時間, ip, 訪問次數)
      .addSink(new JDBCSink()) // 往MySQL插入資料

    env.execute()
  }
}
// 自定義RichSinkFunction往MySQL插入資料
class JDBCSink extends RichSinkFunction[(Long, String, Long)]{
  // 定義連接和前處理器
  var conn:Connection = _
  var insertStatement: PreparedStatement = _

  // 在open函式初始化連接和預編譯器
  override def open(parameters: Configuration): Unit = {
    conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/pv_uv", "root", "123456")
    insertStatement = conn.prepareStatement("insert into unique_visitor value(?, ? ,?)")
  }
  // 在close函式關閉連接和預編譯器
  override def close(): Unit = {
    conn.close()
    insertStatement.close()
  }
  // 在invoke函式指定前處理器的資料和執行插入陳述句
  override def invoke(value: (Long, String, Long), context: SinkFunction.Context[_]): Unit = {
    // 指定預編譯器的資料
    insertStatement.setTimestamp(1, new Timestamp(value._1))
    insertStatement.setString(2, value._2)
    insertStatement.setInt(3, value._3.toInt)
    // 執行預編譯器
    insertStatement.execute()
  }
}

// 基于WindowEnd分組后 在該Process中回傳要插入資料庫的元祖Tuple
class windowEndProcess() extends KeyedProcessFunction[Long, UVWindowEnd, (Long, String, Long)]{
  override def processElement(i: UVWindowEnd, context: KeyedProcessFunction[Long, UVWindowEnd, (Long, String, Long)]#Context, collector: Collector[(Long, String, Long)]): Unit = {
    // 回傳(視窗結束時間, 頁面路徑, 訪問次數)
    collector.collect((i.WindowEnd, i.url, i.Count))
  }
}

// 基于url分組并開窗后 在該Process中統計UV值
class CountUVProcess() extends ProcessWindowFunction[UVItem, UVWindowEnd, String, TimeWindow]{
  override def process(key: String, context: Context, elements: Iterable[UVItem], out: Collector[UVWindowEnd]): Unit = {
    // 用Set集合可以去重的特性 一個ip計為一次訪問
    var userIpSet = Set[String]()
    for(item <- elements){
      userIpSet += item.ip
    }
    // 回傳(訪問的url, 視窗結束時間, 訪問次數)
    out.collect(UVWindowEnd(key, context.window.getEnd, userIpSet.size))
  }
}

MySQL創建表

插入資料后

依賴

<dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.11</artifactId>
            <version>1.10.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.11</artifactId>
            <version>1.10.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
            <version>1.10.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.bahir</groupId>
            <artifactId>flink-connector-redis_2.11</artifactId>
            <version>1.0</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>8.0.25</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-statebackend-rocksdb_2.12</artifactId>
            <version>1.10.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner_2.11</artifactId>
            <version>1.10.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_2.11</artifactId>
            <version>1.10.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-scala-bridge_2.11</artifactId>
            <version>1.10.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-csv</artifactId>
            <version>1.10.1</version>
        </dependency>

    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.4.6</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-assembly-plugin</artifactId>
                <version>3.0.0</version>
                <configuration>
                    <descriptorRefs>
                        <descriptorRef>jar-with-dependencies</descriptorRef>
                    </descriptorRefs>
                </configuration>
                <executions>
                    <execution>
                        <id>make-assembly</id>
                        <phase>package</phase>
                        <goals>
                            <goal>single</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>