<style></style>

Flink-測驗用的fake溫度傳感器

 

Flink中，測驗時，會用到自定義的source，

下為一例，， 該例使用溫度傳感器的格式演示fake日志資料源，

代碼用Scala寫的，

 

傳感器...

 

• 傳感器 - 樣例類

  SensorReads.scala：

        ?x         1

  package sr

  2

  ?

  3

  /**

  4

   * 

  5

   */

  6

  case class SensorReads(id:String,

  7

                         timestap:Long,

  8

                         tempture:Double)

     

 

• 傳感器 - 資料源模擬

  SnsorSrc_4096T.scala：

        46         1

  package sr

  2

  ?

  3

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

  4

  import scala.util.Random

  5

  ?

  6

  /**

  7

   * period, is 4096 millis.

  8

   */

  9

  case class SnsorSrc_4096T extends SourceFunction[SensorReads] {

  10
  11

      var isInRunning: Boolean = true

  12
  13

      ////

  14

      override def run(sourceContext: SourceFunction.SourceContext[

  15

              SensorReads]): Unit = {

  16
  17
  18

          val rand: Random = new Random

  19
  20

          var tptNow4 =

  21

              (1 to 4).map(

  22

                  "snsor_" + _.toString -> (23 + 16 * rand.nextGaussian))

  23

  ?

  24
  25
  26

          while (isInRunning) {

  27

              tptNow4 = tptNow4.map(

  28

                  t => t._1 -> (t._2 + rand.nextGaussian))

  29
  30
  31

              val timeStampNow: Long = System.currentTimeMillis

  32
  33

              tptNow4.foreach{

  34

                  t =>

  35

                      sourceContext.collect( // O.U.T

  36

                          SensorReads(t._1, timeStampNow, t._2) )

  37

                  Thread.sleep(512)  }

  38

              //not set, is stm

  39
  40

              Thread.sleep(2048)  }

  41
  42

      }

  43
  44

      override def cancel(): Unit = isInRunning = false

  45
  46

  }

     

 

 

測驗

 

SnsrSrcAappli.scala：

      13         1

package applis

2

?

3

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

4

import sr._

5

?

6

object SnsrSrcAappli extends App{

7

    val env = StreamExecutionEnvironment.getExecutionEnvironment

8

9

    env.addSource(SnsorSrc_4096T() )

10

                    .print("aaa")

11

12

    env.execute()

13

}

   

 

資料源模擬用case-class，此處使用則可以不寫new，

 

輸出

 

IDEA控制臺上run：

      17         1

log4j:WARN No appenders could be found for logger (org.apache.flink.api.scala.ClosureCleaner$).

2

log4j:WARN Please initialize the log4j system properly.

3

log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.

4

aaa:3> SensorReads(snsor_1,1573556705508,30.383394411578916)

5

aaa:4> SensorReads(snsor_2,1573556705508,21.397405872448672)

6

aaa:5> SensorReads(snsor_3,1573556705508,20.598086139457727)

7

aaa:6> SensorReads(snsor_4,1573556705508,18.30066983735531)

8

aaa:7> SensorReads(snsor_1,1573556709627,30.120955223032546)

9

aaa:8> SensorReads(snsor_2,1573556709627,22.38746867201145)

10

aaa:1> SensorReads(snsor_3,1573556709627,20.45357507067989)

11

aaa:2> SensorReads(snsor_4,1573556709627,17.18467261133715)

12

aaa:3> SensorReads(snsor_1,1573556713729,31.686487593592904)

13

aaa:4> SensorReads(snsor_2,1573556713729,20.67106361911623)

14

aaa:5> SensorReads(snsor_3,1573556713729,21.27724215221553)

15

aaa:6> SensorReads(snsor_4,1573556713729,16.84273306583804)

16

?

17

Process finished with exit code -1

   

 

...

如果SnsorSrc_4096T.scala中，「當前溫度」.foreach這樣寫：

      5         1

tptNow4.foreach{

2

    t =>

3

        sourceContext.collect( // O.U.T

4

            SensorReads(t._1, System.currentTimeMillis, t._2) )

5

    Thread.sleep(512)  }

   

 

那么結果就會是：

      25         1

log4j:WARN No appenders could be found for logger (org.apache.flink.api.scala.ClosureCleaner$).

2

log4j:WARN Please initialize the log4j system properly.

3

log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.

4

aaa:5> SensorReads(snsor_1,1573561932216,20.427373784204445)

5

aaa:6> SensorReads(snsor_2,1573561932739,19.043151948599565)

6

aaa:7> SensorReads(snsor_3,1573561933251,16.506314894849734)

7

aaa:8> SensorReads(snsor_4,1573561933764,42.18791135873409)

8

aaa:1> SensorReads(snsor_1,1573561936326,20.216273863226476)

9

aaa:2> SensorReads(snsor_2,1573561936838,19.77488458362011)

10

aaa:3> SensorReads(snsor_3,1573561937351,17.49661332626548)

11

aaa:4> SensorReads(snsor_4,1573561937864,42.37076203420432)

12

aaa:5> SensorReads(snsor_1,1573561940425,19.582646754534526)

13

aaa:6> SensorReads(snsor_2,1573561940938,18.148182987020572)

14

aaa:7> SensorReads(snsor_3,1573561941451,17.028248074961432)

15

aaa:8> SensorReads(snsor_4,1573561941963,42.969281620777075)

16

aaa:1> SensorReads(snsor_1,1573561944525,20.659855873131406)

17

aaa:2> SensorReads(snsor_2,1573561945038,19.437515708059177)

18

aaa:3> SensorReads(snsor_3,1573561945550,18.336847248220565)

19

aaa:4> SensorReads(snsor_4,1573561946063,43.58727112744526)

20

aaa:5> SensorReads(snsor_1,1573561948625,19.317498008380674)

21

aaa:6> SensorReads(snsor_2,1573561949137,21.86602577501872)

22

aaa:7> SensorReads(snsor_3,1573561949650,19.109322091177216)

23

aaa:8> SensorReads(snsor_4,1573561950163,43.48043890977487)

24

?

25

Process finished with exit code -1

   

這樣一來事件時間就都不一樣了， 可根據需要模擬的情況改動....

轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/46545.html

標籤：大數據

上一篇：Elastic search集群新增節點（同一集群，同一物理機）

下一篇：辭職了，何去何從呢