ApacheFlink:如何解释datastream.print输出?

cbeh67ev  于 2021-06-24  发布在  Flink
关注(0)|答案(1)|浏览(396)

我是一个新手,在Flink,试图了解如何使用它是最有效的。
我正在尝试使用windowapi,从csv文件读取。读取的行被转换成case类,因此

case class IncomingDataUnit (
sensorUUID: String, radiationLevel: Int,photoSensor: Float,
humidity: Float,timeStamp: Long, ambientTemperature: Float)
  extends Serializable {

}

我是这样读的:

env.readTextFile(inputPath).map(datum => {
      val fields = datum.split(",")
      IncomingDataUnit(
        fields(0),              // sensorUUID
        fields(1).toInt,        // radiationLevel
        fields(2).toFloat,      // photoSensor
        fields(3).toFloat,      // humidity
        fields(4).toLong,       // timeStamp
        fields(5).toFloat       // ambientTemperature
      )
    })

稍后,使用一个简单的窗口,我尝试打印最高环境温度,因此:

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)

val readings =
      readIncomingReadings(env,"./sampleIOTTiny.csv")
      .map(e => (e.sensorUUID,e.ambientTemperature))
      .timeWindowAll(Time.of(5,TimeUnit.MILLISECONDS))
      .trigger(CountTrigger.of(5))
      .evictor(CountEvictor.of(4))
      .max(1)

readings.print

输出包含以下内容(来自一组调试日志语句):

1> (probe-987f2cb6,29.43)
1> (probe-987f2cb6,29.43)
3> (probe-dccefede,30.02)
3> (probe-42a9ddca,22.07)
2> (probe-df2d4cad,22.87)
2> (probe-20c609fb,27.62)
4> (probe-dccefede,30.02)

我想了解的是,人们是如何理解这一点的?重复的1>s代表什么?
同样让我困惑的是,probe-987f2cb6与我数据集中的环境温度29.43不符。它对应一个不同的值(确切地说是14.72)。
以下数据集仅供参考:

probe-f076c2b0,201,842.53,75.5372,1448028160,29.37
probe-dccefede,199,749.25,78.6057,1448028160,27.46
probe-f29f9662,199,821.81,81.7831,1448028160,22.35
probe-5dac1d9f,195,870.71,83.1028,1448028160,15.98
probe-6c75cfbe,198,830.06,82.5607,1448028160,30.02
probe-4d78b545,204,778.42,78.412,1448028160,25.92
probe-400c5cdf,204,711.65,73.585,1448028160,22.18
probe-df2d4cad,199,820.8,72.936,1448028161,16.18
probe-f4ef109e,199,785.68,77.5647,1448028161,16.36
probe-3fac3350,200,720.12,78.2073,1448028161,19.19
probe-42a9ddca,193,819.12,74.3712,1448028161,22.07
probe-252a5bbd,197,710.32,80.6072,1448028161,14.64
probe-987f2cb6,200,750.4,76.0533,1448028161,14.72
probe-24444323,197,816.06,84.0816,1448028161,4.405
probe-6dd6fdc4,201,717.64,78.4031,1448028161,29.43
probe-20c609fb,204,804.37,84.5243,1448028161,22.87
probe-c027fdc9,195,858.61,81.7682,1448028161,24.47
probe-2c6cd3de,198,826.96,85.26,1448028162,18.99
probe-960906ca,197,797.63,77.4359,1448028162,27.62

我可能错过了很多。请戳我一下。
忘了提一下:我使用的是flink0.10.0。

lrl1mhuk

lrl1mhuk1#

“>x”表示打印结果元组的并行任务的任务id。我只是想知道为什么输出显示值1到4——因为您使用的是非并行窗口(数据流不是通过 .keyBy() )我也希望指纹是连锁的和不平行的。但可能不是,4个并行打印任务正在运行。
关于您的结果:如果窗口触发,则对所有元组计算max over字段1,并对字段0使用窗口的head元组值。如果要返回包含最大值的完整元组,可以使用 maxBy() 而不是 max() .

相关问题