(15) Flick custom source

List of articles

• • 
    
• ​ ​source​​
• ​ ​ The main class ​​
  • 
    

source

Log in and copy

       

        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

Log in and copy

       

        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.