带有自由单子的文件 I/O

Question

我有一个CSV文件，我需要解析并对每条记录执行一些操作。我如何使用免费Monads？目前，我正在将整个文件加载到内存中，并想知道是否有更好的解决方案。以下是我的程序：

for {
    reader <- F.getReader("my_file.csv")
    csvRecords <- C.readCSV(reader)
    _ <- I.processCSV(csvRecords)
    _ <- F.close(reader)
} yield()

此代码适用于较小的文件，但是如果我有非常大的文件(超过 1 GB(，这将不能很好地工作。我正在使用Commons CSV来阅读CSVRecords。

Answer 1

查看您的要点代码，我认为带有注释的行正是您根本不想要的行：

  object CSVIOInterpreter extends (CSVIO ~> Future) {
    import scala.collection.JavaConverters._
    override def apply[A](fa: CSVIO[A]): Future[A] = fa match {
      case ReadCSV(reader) => Future.fromTry(Try {
        CSVFormat.RFC4180
          .withFirstRecordAsHeader()
          .parse(reader)
          .getRecords // Loads the complete file
          .iterator().asScala.toStream
      })
    }
  }

只需删除整条getRecords行即可。 CSVFormat.parse返回一个已经实现Iterable<CSVRecord>的 CSVParser 实例。getRecords调用是唯一强制它读取整个文件的东西。

实际上你可以看到CSVParser.getRecords实现，它是

 public List<CSVRecord> getRecords() throws IOException {
     CSVRecord rec;
     final List<CSVRecord> records = new ArrayList<>();
     while ((rec = this.nextRecord()) != null) {
         records.add(rec);
     }
     return records;
 }

因此，它只是使用调用来实现整个文件this.nextRecord这显然是 API 中更"核心"的部分。

因此，当我在没有getRecords调用的情况下执行简化版本的代码时：

import cats._
import cats.free.Free
import java.io._
import org.apache.commons.csv._
import scala.collection.JavaConverters._
trait Action[A] {
  def run(): A
}
object F {
  import Free.liftF
  case class GetReader(fileName: String) extends Action[Reader] {
    override def run(): Reader = new FileReader(fileName)
  }
  case class CloseReader(reader: Reader) extends Action[Unit] {
    override def run(): Unit = reader.close()
  }
  def getReader(fileName: String): Free[Action, Reader] = liftF(GetReader(fileName))
  def close(reader: Reader): Free[Action, Unit] = liftF(CloseReader(reader))
}
object C {
  import Free.liftF
  case class ReadCSV(reader: Reader) extends Action[CSVParser] {
    override def run(): CSVParser = CSVFormat.DEFAULT.parse(reader)
  }
  def readCSV(reader: Reader): Free[Action, CSVParser] = liftF(ReadCSV(reader))
}
object I {
  import Free.liftF
  case class ProcessCSV(parser: CSVParser) extends Action[Unit] {
    override def run(): Unit = {
      for (r <- parser.asScala)
        println(r)
    }
  }
  def processCSV(parser: CSVParser): Free[Action, Unit] = liftF(ProcessCSV(parser))
}
object Runner {
  import cats.arrow.FunctionK
  import cats.{Id, ~>}
  val runner = new (Action ~> Id) {
    def apply[A](fa: Action[A]): Id[A] = fa.run()
  }
  def run[A](free: Free[Action, A]): A = {
    free.foldMap(runner)
  }
}

def test() = {
  val free = for {
    //        reader <- F.getReader("my_file.csv")
    reader <- F.getReader("AssetsImportCompleteSample.csv")
    csvRecords <- C.readCSV(reader)
    _ <- I.processCSV(csvRecords)
    _ <- F.close(reader)
  } yield ()
  Runner.run(free)
}

它似乎在逐行模式下工作正常。

Answer 2

在这里，

我如何使用 CSV 文件读取并对其进行一些操作 -我使用scala.io.Source.fromFile()

我创建了一个 CSV 文件header类型的case class，以使数据更易于访问和操作。

PS：我对monads一无所知，而且我是Scala的初学者。我发布了这个，因为它可能会有所帮助。

case class AirportData(id:Int, ident:String, name:String, typeAirport:String, latitude_deg:Double,
longitude_deg:Double, elevation_ft:Double, continent:String, iso_country:String, iso_region:String,
municipality:String)
object AirportData extends App {
def toDoubleOrNeg(s: String): Double = {
  try {
    s.toDouble
   } catch {
    case _: NumberFormatException => -1 
   }
 }
val source = scala.io.Source.fromFile("resources/airportData/airports.csv")
val lines = source.getLines().drop(1)
val data = lines.flatMap { line =>
val p = line.split(",")
  Seq(AirportData(p(0).toInt, p(1).toString, p(2).toString, p(3).toString, toDoubleOrNeg(p(4)), toDoubleOrNeg(p(5)), 
      toDoubleOrNeg(p(6)), p(7).toString, p(8).toString, p(9).toString, p(10).toString))
 }.toArray   
 source.close()
 println(data.length)
 data.take(10) foreach println
}