[Flink]测试用的fake温度传感器
Flink-测试用的fake温度传感器
Flink中,测试时,会用到自定义的source。
下为一例。。 该例使用温度传感器的格式演示fake日志数据源。
代码用Scala写的。
传感器...
传感器 - 样例类
SensorReads.scala
:x1package sr
2
3/**
4*
5*/
6case class SensorReads(id:String,
7timestap:Long,
8tempture:Double)
传感器 - 数据源模拟
SnsorSrc_4096T.scala
:461package sr
2
3import org.apache.flink.streaming.api.functions.source.SourceFunction
4import scala.util.Random
5
6/**
7* period, is 4096 millis.
8*/
9case class SnsorSrc_4096T extends SourceFunction[SensorReads] {
1011var isInRunning: Boolean = true
1213////
14override def run(sourceContext: SourceFunction.SourceContext[
15SensorReads]): Unit = {
161718val rand: Random = new Random
1920var tptNow4 =
21(1 to 4).map(
22"snsor_" + _.toString -> (23 + 16 * rand.nextGaussian))
23
242526while (isInRunning) {
27tptNow4 = tptNow4.map(
28t => t._1 -> (t._2 + rand.nextGaussian))
293031val timeStampNow: Long = System.currentTimeMillis
3233tptNow4.foreach{
34t =>
35sourceContext.collect( // O.U.T
36SensorReads(t._1, timeStampNow, t._2) )
37Thread.sleep(512) }
38//not set, is stm
3940Thread.sleep(2048) }
4142}
4344override def cancel(): Unit = isInRunning = false
4546}
测试
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
这样一来事件时间就都不一样了。 可根据需要模拟的情况改动....
赞 (0)