pv-current/deviation/app.scala

import org.apache.spark.sql.types.{StructType, StructField
                                  ,StringType, TimestampType
                                  ,DoubleType}
import org.apache.spark.sql.functions.{concat, substring}

val START_TIME = 7
val END_TIME = 17
val TF = 30
val SAMPLE_INT = 5 // 采用频率，目前规定为5min
val I0 = 1
val IM = 12
val STATION_ID = "STATION_ID"
val DEVICE_ID = "DEVICE_ID"
val TIME_HEADER = "MONITOR_TIME"
val REPORT_TIME = "ALERT_TIME"
val V_HEADER = "VOLTAGE_VALUE"
val CURRENT_HEADER = "CURRENT_VALUE"
val GRP_ID = "grp_id"
val STR_HEADER = "str_no"
val DEVIATION = "DEVIATION"    // 输出结果中 离散率 列名
val STR_STATUS = "STR_STATUS"  // 输出结果中 组串状态 列名
val COM_FAIL = -1
val UNIT_INVALID = -2
val STR_INVALID = -3

val dfSchema = StructType(Array(
   StructField(STATION_ID, StringType, true)
  ,StructField(DEVICE_ID, StringType, true)
  ,StructField(TIME_HEADER, TimestampType, true)
  ,StructField(V_HEADER, DoubleType, true)
  ,StructField(CURRENT_HEADER, DoubleType, true)
  ))

val df = spark.read.format("csv").option("header", "true").schema(dfSchema).load("comb1122.csv")
val df2 = df.filter(hour($"MONITOR_TIME") >= START_TIME && hour($"MONITOR_TIME") <= END_TIME)
val df3 = df2.withColumn(GRP_ID, substring(col(DEVICE_ID), 0, 6)).withColumn(
                         GRP_ID, concat(col(STATION_ID), col(GRP_ID)))
val df4 = df3.withColumn(STR_HEADER, substring(col(DEVICE_ID), 7, 2))
val unit_grps = df4.groupBy(GRP_ID)
val res = unit_grps.apply(unit_deviation)