SplitData.scala 2.0 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071
  1. package com.persagy.iot.func
  2. import com.persagy.iot.app.IOTApp.no_value
  3. import com.persagy.iot.bean.IOTData
  4. import com.persagy.iot.utils.IOTUtils.timestampConverter
  5. import org.apache.flink.api.common.functions.FlatMapFunction
  6. import org.apache.flink.api.common.time.Time
  7. import org.apache.flink.util.Collector
  8. /**
  9. * 由于每一条数据包含多个功能号,表号
  10. * 自定义 flatMap 函数
  11. */
  12. class SplitData extends FlatMapFunction[String, IOTData] {
  13. override def flatMap(input: String, collector: Collector[IOTData]): Unit = {
  14. try {
  15. /** 处理数据,提取 数据上报包 */
  16. val arr1 = input.split("\t")
  17. val builds: Array[String] = arr1(3).split("&")
  18. for (elem <- builds) {
  19. val arr2: Array[String] = elem.split(";")
  20. val build: String = arr2(0).toString
  21. val sign: String = arr2(5).toString
  22. val eventTimeStr: String = arr2(3)
  23. val eventTime: Long = timestampConverter("yyyyMMddHHmmss", eventTimeStr, Time.hours(8).toMilliseconds)
  24. val sysTime: Long = System.currentTimeMillis()
  25. for (i <- 7 until (arr2.length, 2) if (!no_value.equals(arr2(i + 1)))) {
  26. val funId: String = arr2(i)
  27. val value: Double = arr2(i + 1).toDouble
  28. val status: Int = getStatus(eventTime, sysTime)
  29. /** rowKey */
  30. val rowKey = build + ":" + sign + ":" + funId + ":" + eventTime
  31. /** 转为 iotData 实体类 */
  32. val iotData: IOTData = IOTData(rowKey, build, sign, funId, value, eventTimeStr, eventTime, sysTime, 0L, status)
  33. collector.collect(iotData)
  34. }
  35. }
  36. } catch {
  37. case ex: Exception => return
  38. }
  39. }
  40. /**
  41. * 判断该数据是正常,迟到,丢数
  42. * @param eventTime 数据产生时间
  43. * @param sysTime 系统时间
  44. */
  45. def getStatus(eventTime: Long, sysTime: Long): Int ={
  46. val var1: Long = sysTime - eventTime
  47. if (var1 <= 1800){
  48. 0
  49. } else if (1800 < var1 && var1 <= 3600) {
  50. 1
  51. } else {
  52. 3
  53. }
  54. }
  55. }