Scala-akka: - Передача аргументов командной строки в sbt run для поиска файлов .csv

Обе ваши команды sbt (то есть sbt "run data" и sbt "run src/main/scala/data") выглядят правильно, при условии, что вы запускаете sbt из корня проекта Scala с исходным кодом в «src/main/scala/» и csv-файлами в «src/main/scala». /данные/".

Пара наблюдаемых проблем с кодом:

При создании fileSource есть большая вероятность, что new File().listFiles() получит больше файлов, чем вы собираетесь включить (например, файлы, отличные от CSV, скрытые файлы и т. д.), в результате чего после прохождения через Framing.delimiter() появится один большой двоичный объект, который впоследствии будет удален drop(1). В таком случае карта «sensors» будет пустой, в результате чего на выходе будут все 0.

Мне удалось воспроизвести результат «все 0», используя ваш точный исходный код и «build.sbt», по-видимому, из-за файлов, отличных от CSV (в моем тестовом примере, файл «.DS_Store»), включенных в listFiles().

Предоставление конкретных критериев выбора файлов для listFiles(), таких как включение только «*.csv», как показано ниже, должно решить проблему:

val fileSource = Source.fromIterator( () =>
  new File(directoryPath).listFiles((_, name) => name.endsWith(".csv")).iterator
)

Другая проблема заключается в том, что логика вычисления (humidityData.sum / humidityData.count) для min и max неверна, по существу повторяя вычисление avg. Чтобы их вычислить, можно расширить параметры в HumidityData следующим образом:

case class HumidityData(sum: Double, count: Int, min: Double, max: Double) {...}

Затем min/max можно обновить примерно так:

humidity match {
  case Some(h) =>
    sensors.put(sensorId, sensors.getOrElse(sensorId, HumidityData(0.0, 0, Double.MaxValue, 0.0)) match {
      case HumidityData(sum, count, min, max) =>
        HumidityData(sum + h, count + 1, Math.min(h, min), Math.max(h, max))
    })
  case None =>
    failedMeasurements += 1
}

В качестве примечания я бы рекомендовал отделить данные от кода, переместив файлы данных из «src/main/scala/» и, возможно, поместив их, скажем, в «src/main/resources/data/».

Тестирование со следующими файлами данных csv...

File src/main/resources/data/sensor_data1.csv:
sensor-id,humidity
s1,80
s3,NaN
s2,78
s1,98

File src/main/resources/data/sensor_data2.csv:
sensor-id,humidity
s1,70
s3,80
s2,60

$ sbt "run src/main/resources/data"
[info] welcome to sbt 1.5.5 (Oracle Corporation Java 1.8.0_181)
[info] loading settings for project global-plugins from idea.sbt ...
[info] loading global plugins from /Users/leo/.sbt/1.0/plugins
[info] loading project definition from /Users/leo/work/so-75459442/project
[info] loading settings for project root from build.sbt ...
[info] set current project to sensor-task (in build file:/Users/leo/work/so-75459442/)
[info] running HumiditySensorStatistics src/main/resources/data
Looking for CSV files in directory: src/main/resources/data
Num of processed sensors: 3
Num of processed measurements: 7
Num of failed measurements: 1
Sensors with highest avg humidity:
sensor-id,min,avg,max
s3,NaN,NaN,NaN
s1,70.0,82.66666666666667,98.0
s2,60.0,69.0,78.0
[success] Total time: 2 s, completed Feb 16, 2023 10:33:35 AM

Прилагается исправленный исходный код.

import java.io.File
import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.{FileIO, Framing, Sink, Source}
import akka.util.ByteString
import scala.collection.mutable
import scala.concurrent.ExecutionContext.Implicits.global

object HumiditySensorStatistics {

  case class HumidityData(sum: Double, count: Int, min: Double, max: Double) {
    def avg: Option[Double] = if (count > 0) Some(sum / count) else None
  }

  case class SensorStats(min: Option[Double], avg: Option[Double], max: Option[Double])

  def main(args: Array[String]): Unit = {
    val directoryPath = args(0)
    implicit val system: ActorSystem = ActorSystem("HumiditySensorStatistics")
    // implicit val materializer: ActorMaterializer = ActorMaterializer()  // Not needed for Akka Stream 2.6+
    val sensors = mutable.Map[String, HumidityData]()
    var failedMeasurements = 0

    println(s"Looking for CSV files in directory: $directoryPath")

    val fileSource = Source.fromIterator( () =>
      new File(directoryPath).listFiles((_, name) => name.endsWith(".csv")).iterator
    )
    val measurementSource = fileSource.flatMapConcat(f => FileIO.fromPath(f.toPath))
      .via(Framing.delimiter(ByteString("\n"), maximumFrameLength = 1024, allowTruncation = true))
      .drop(1) // skip header line
      .map(_.utf8String)
      .map(line => {
        val fields = line.split(",")
        (fields(0), fields(1))
      })
    val sink = Sink.foreach[(String, String)](data => {
      val sensorId = data._1
      val humidity = data._2.toDoubleOption

      humidity match {
        case Some(h) =>
          sensors.put(sensorId, sensors.getOrElse(sensorId, HumidityData(0.0, 0, Double.MaxValue, 0.0)) match {
            case HumidityData(sum, count, min, max) =>
              HumidityData(sum + h, count + 1, Math.min(h, min), Math.max(h, max))
          })
        case None =>
          failedMeasurements += 1
      }
    })

    measurementSource.runWith(sink).onComplete(_ => {
      val numSensorsProcessed = sensors.size
      val numMeasurementsProcessed = sensors.values.map(_.count).sum
      val numFailedMeasurements = failedMeasurements

      println(s"Num of processed sensors: $numSensorsProcessed")
      println(s"Num of processed measurements: $numMeasurementsProcessed")
      println(s"Num of failed measurements: $numFailedMeasurements")

      val statsBySensor = sensors.map {
        case (sensorId, humidityData) =>
          val stats = SensorStats(
            min = Some(humidityData.min),
            avg = humidityData.avg,
            max = Some(humidityData.max)
          )
          (sensorId, stats)
      }

      println("Sensors with highest avg humidity:")
      println("sensor-id,min,avg,max")
      statsBySensor.toList.sortBy(_._2.avg).reverse.foreach {
        case (sensorId, stats) =>
          println(s"$sensorId,${stats.min.getOrElse("NaN")},${stats.avg.getOrElse("NaN")},${stats.max.getOrElse("NaN")}")
      }

      system.terminate()
    })
  }
}