| 1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071 |
- package com.persagy.iot.func
- import com.persagy.iot.app.IOTApp.no_value
- import com.persagy.iot.bean.IOTData
- import com.persagy.iot.utils.IOTUtils.timestampConverter
- import org.apache.flink.api.common.functions.FlatMapFunction
- import org.apache.flink.api.common.time.Time
- import org.apache.flink.util.Collector
- /**
- * 由于每一条数据包含多个功能号,表号
- * 自定义 flatMap 函数
- */
- class SplitData extends FlatMapFunction[String, IOTData] {
- override def flatMap(input: String, collector: Collector[IOTData]): Unit = {
- try {
- /** 处理数据,提取 数据上报包 */
- val arr1 = input.split("\t")
- val builds: Array[String] = arr1(3).split("&")
- for (elem <- builds) {
- val arr2: Array[String] = elem.split(";")
- val build: String = arr2(0).toString
- val sign: String = arr2(5).toString
- val eventTimeStr: String = arr2(3)
- val eventTime: Long = timestampConverter("yyyyMMddHHmmss", eventTimeStr, Time.hours(8).toMilliseconds)
- val sysTime: Long = System.currentTimeMillis()
- for (i <- 7 until (arr2.length, 2) if (!no_value.equals(arr2(i + 1)))) {
- val funId: String = arr2(i)
- val value: Double = arr2(i + 1).toDouble
- val status: Int = getStatus(eventTime, sysTime)
- /** rowKey */
- val rowKey = build + ":" + sign + ":" + funId + ":" + eventTime
- /** 转为 iotData 实体类 */
- val iotData: IOTData = IOTData(rowKey, build, sign, funId, value, eventTimeStr, eventTime, sysTime, 0L, status)
- collector.collect(iotData)
- }
- }
- } catch {
- case ex: Exception => return
- }
- }
- /**
- * 判断该数据是正常,迟到,丢数
- * @param eventTime 数据产生时间
- * @param sysTime 系统时间
- */
- def getStatus(eventTime: Long, sysTime: Long): Int ={
- val var1: Long = sysTime - eventTime
- if (var1 <= 1800){
- 0
- } else if (1800 < var1 && var1 <= 3600) {
- 1
- } else {
- 3
- }
- }
- }
|