DataQualityConcept.scala

readCsv("/mnt/data/110218.csv", 'datetime) { row =>
  saveToDb(telemetryConnect, sinkSchema)(row) {
    outliersToAvg(5.second)('speed, 'oilPomp, 'KMPos) ~>
    enrichWith(weather) ~>
    interpolate('speed)
  } ~
  persistRule(rulesConnect, sinkSchema)(row) & alert(alertConnect, schema)(row) {
    Assert('PosKM.as[Double] > 10).timed(max=10.seconds)
  } ~
  persistRollUp(rollUpsConnect, sinkSchema)(row) { 
    max('speed) as "Максимальная скорость" ~>
    distinct('loco_id) as "Уникальные составы"
  }
}