当前位置：网站首页>(15) Flick custom source

(15) Flick custom source

2022-07-02 07:48:00 【wx5ba7ab4695f27】

List of articles

source
The main class

source

       
        package com.htsec.test
        

        
import org.apache.flink.streaming.api.functions.source.SourceFunction
        

        
import scala.util.Random
        

        
class MySource() extends SourceFunction[SersorReading] {
        
  //flag  Indicates whether the data source is running normally 
        
  var running: Boolean = true
        

        
  override def run(ctx: SourceFunction.SourceContext[SersorReading]): Unit = {
        
    // Initialize a random number generator 
        
    val random = new Random()
        
    // Random generation 10 Initial temperature value of sensors , And sensors id Form a binary 
        
    var curTemp = 1.to(10).map(
        
      i => ("sensor_" + i, random.nextGaussian() * 20 + 60)
        
    )
        
    while (running) {
        
      // Update based on last temperature value 
        
      curTemp.map(
        
        data => (data._1, data._2 + random.nextGaussian())
        
      )
        
      // Get the current timestamp 
        
      val curTime = System.currentTimeMillis()
        
      curTemp.foreach(
        
        data => ctx.collect(SersorReading(data._1,curTime,data._2))
        
      )
        
    }
        
  }
        

        
  override def cancel(): Unit = {
        
    running = false
        
  }
        
}
       
       
        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.

The main class

       
        package com.htsec.test
        

        
import org.apache.flink.streaming.api.scala._
        

        
case class SersorReading(id: String,timeStamp:Long,temperature : Double)
        

        
object SourceTest {
        
  def main(args: Array[String]): Unit = {
        
    val env = StreamExecutionEnvironment.getExecutionEnvironment
        
    val lines = env.addSource(new MySource)
        
    lines.print()
        
    env.execute()
        
  }
        
}
       
       
        1.
        2.
        3.
        4.
        5.
        6.
        7.
        8.
        9.
        10.
        11.
        12.
        13.
        14.