Как прочитать большой CSV файл с классом Scala Stream?

Ответ 1

Просто используйте Source.fromFile(...).getLines, как вы уже сказали.

Это возвращает Итератор, который уже ленив (вы использовали бы поток как ленивую коллекцию, в которой вы хотели бы сохранить ранее полученные значения, чтобы их можно было прочитать снова)

Если у вас проблемы с памятью, проблема будет заключаться в том, что вы делаете после getLines. Любая операция типа toList, которая создает строгий сбор, вызовет проблему.

Ответ 2

Надеюсь, вы не имеете в виду Scala collection.immutable.Stream с Stream. Это не то, что вы хотите. Поток ленив, но делает memoization.

Я не знаю, что вы планируете делать, но просто чтение файла по строкам должно работать очень хорошо, не используя большие объемы памяти.

getLines должен оцениваться лениво и не должен вылетать (если в вашем файле не более 2-3 строк, afaik). Если это так, спросите на # scala или напишите билет с ошибкой (или выполните оба варианта).

Ответ 3

Если вы хотите обработать большой файл по очереди, избегая при этом необходимости полностью загружать содержимое всего файла в память, вы можете использовать Iterator, возвращенный scala.io.Source.

У меня есть небольшая функция, tryProcessSource, (содержащая две подфункции), которые я использую для этих типов прецедентов. Функция принимает до четырех параметров, из которых требуется только первая. Другие параметры имеют нормальные значения по умолчанию.

Здесь профиль функции (реализация полной функции находится внизу):

def tryProcessSource(
  file: File,
  parseLine: (Int, String) => Option[List[String]] =
    (index, unparsedLine) => Some(List(unparsedLine)),
  filterLine: (Int, List[String]) => Option[Boolean] =
    (index, parsedValues) => Some(true),
  retainValues: (Int, List[String]) => Option[List[String]] =
    (index, parsedValues) => Some(parsedValues),
): Try[List[List[String]]] = {
  ???
}

Требуется первый параметр file: File. И это всего лишь любой действительный экземпляр java.io.File, который указывает на текстовый текстовый файл, такой как CSV.

Второй параметр parseLine: (Int, String) => Option[List[String]] не является обязательным. И если это предусмотрено, это должна быть функция, ожидающая получить два входных параметра; index: Int, unparsedLine: String. А затем верните Option[List[String]]. Функция может возвращать Some wrapped List[String], состоящий из допустимых значений столбца. Или он может вернуть None, который указывает, что весь процесс потоковой передачи прерывается раньше. Если этот параметр не указан, предоставляется значение по умолчанию (index, line) => Some(List(line)). Это значение по умолчанию приводит к тому, что вся строка возвращается как одно значение String.

Третий параметр filterLine: (Int, List[String]) => Option[Boolean] не является обязательным. И если это предусмотрено, это должна быть функция, ожидающая получить два входных параметра; index: Int, parsedValues: List[String]. А затем верните Option[Boolean]. Функция может возвращать Some wrapped Boolean, указывающий, должна ли эта конкретная строка быть включена в выход. Или он может вернуть None, который указывает, что весь процесс потоковой передачи прерывается раньше. Если этот параметр не указан, предоставляется значение по умолчанию (index, values) => Some(true). Это значение по умолчанию приводит к включению всех строк.

Четвертый и последний параметр retainValues: (Int, List[String]) => Option[List[String]] не является обязательным. И если это предусмотрено, это должна быть функция, ожидающая получить два входных параметра; index: Int, parsedValues: List[String]. А затем верните Option[List[String]]. Функция может возвращать Some wrapped List[String], состоящий из некоторого подмножества и/или изменения существующих значений столбца. Или он может вернуть None, который указывает, что весь процесс потоковой передачи прерывается раньше. Если этот параметр не указан, предоставляется значение по умолчанию (index, values) => Some(values). Это значение по умолчанию приводит к значениям, проанализированным вторым параметром, parseLine.

Рассмотрим файл со следующим содержимым (4 строки):

street,street2,city,state,zip
100 Main Str,,Irving,TX,75039
231 Park Ave,,Irving,TX,75039
1400 Beltline Rd,Apt 312,Dallas,Tx,75240

Следующий вызывающий профиль...

val tryLinesDefaults =
  tryProcessSource(new File("path/to/file.csv"))

... приводит к этому выводу для tryLinesDefaults (неизмененное содержимое файла):

Success(
  List(
    List("street,street2,city,state,zip"),
    List("100 Main Str,,Irving,TX,75039"),
    List("231 Park Ave,,Irving,TX,75039"),
    List("1400 Beltline Rd,Apt 312,Dallas,Tx,75240")
  )
)

Следующий вызывающий профиль...

val tryLinesParseOnly =
  tryProcessSource(
      new File("path/to/file.csv")
    , parseLine =
        (index, unparsedLine) => Some(unparsedLine.split(",").toList)
  )

... приводит к этому выводу для tryLinesParseOnly (каждая строка анализируется в значениях отдельных столбцов):

Success(
  List(
    List("street","street2","city","state","zip"),
    List("100 Main Str","","Irving,TX","75039"),
    List("231 Park Ave","","Irving","TX","75039"),
    List("1400 Beltline Rd","Apt 312","Dallas","Tx","75240")
  )
)

Следующий вызывающий профиль...

val tryLinesIrvingTxNoHeader =
  tryProcessSource(
      new File("C:/Users/Jim/Desktop/test.csv")
    , parseLine =
        (index, unparsedLine) => Some(unparsedLine.split(",").toList)
    , filterLine =
        (index, parsedValues) =>
          Some(
            (index != 0) && //skip header line
            (parsedValues(2).toLowerCase == "Irving".toLowerCase) && //only Irving
            (parsedValues(3).toLowerCase == "Tx".toLowerCase)
          )
  )

... приводит к этому выводу для tryLinesIrvingTxNoHeader (каждая строка анализируется на отдельные значения столбца, без заголовка и только две строки в Irving, Tx):

Success(
  List(
    List("100 Main Str","","Irving,TX","75039"),
    List("231 Park Ave","","Irving","TX","75039"),
  )
)

Здесь вся реализация функции tryProcessSource:

import scala.io.Source
import scala.util.Try

import java.io.File

def tryProcessSource(
  file: File,
  parseLine: (Int, String) => Option[List[String]] =
    (index, unparsedLine) => Some(List(unparsedLine)),
  filterLine: (Int, List[String]) => Option[Boolean] =
    (index, parsedValues) => Some(true),
  retainValues: (Int, List[String]) => Option[List[String]] =
    (index, parsedValues) => Some(parsedValues)
): Try[List[List[String]]] = {
  def usingSource[S <: Source, R](source: S)(transfer: S => R): Try[R] =
    try {Try(transfer(source))} finally {source.close()}
  def recursive(
    remaining: Iterator[(String, Int)],
    accumulator: List[List[String]],
    isEarlyAbort: Boolean =
      false
  ): List[List[String]] = {
    if (isEarlyAbort || !remaining.hasNext)
      accumulator
    else {
      val (line, index) =
        remaining.next
      parseLine(index, line) match {
        case Some(values) =>
          filterLine(index, values) match {
            case Some(keep) =>
              if (keep)
                retainValues(index, values) match {
                  case Some(valuesNew) =>
                    recursive(remaining, valuesNew :: accumulator) //capture values
                  case None =>
                    recursive(remaining, accumulator, isEarlyAbort = true) //early abort
                }
              else
                recursive(remaining, accumulator) //discard row
            case None =>
              recursive(remaining, accumulator, isEarlyAbort = true) //early abort
          }
        case None =>
          recursive(remaining, accumulator, isEarlyAbort = true) //early abort
      }
    }
  }
  Try(Source.fromFile(file)).flatMap(
    bufferedSource =>
      usingSource(bufferedSource) {
        source =>
          recursive(source.getLines().buffered.zipWithIndex, Nil).reverse
      }
  )
}

Хотя это решение относительно лаконично, мне потребовалось немало времени и много рефакторингов, прежде чем я смог наконец добраться досюда. Пожалуйста, дайте мне знать, видите ли вы какие-либо способы улучшения.

ОБНОВЛЕНИЕ: я только что задал вопрос ниже fooobar.com/info/35057/.... И теперь имеет ответ, фиксирующий ошибку, указанную ниже.

У меня возникла идея попытаться сделать это еще более общим, изменив параметр retainValues на transformLine с новым определением функции generic-ified ниже. Тем не менее, я продолжаю получать ошибку выделения в IntelliJ "Выражение типа Some [List [String]] не соответствует ожидаемому типу Option [A]" и не смогло выяснить, как изменить значение по умолчанию, чтобы ошибка уходит.

def tryProcessSource2[A <: AnyRef](
  file: File,
  parseLine: (Int, String) => Option[List[String]] =
    (index, unparsedLine) => Some(List(unparsedLine)),
  filterLine: (Int, List[String]) => Option[Boolean] =
    (index, parsedValues) => Some(true),
  transformLine: (Int, List[String]) => Option[A] =
    (index, parsedValues) => Some(parsedValues)
): Try[List[A]] = {
  ???
}

Любая помощь в том, как сделать эту работу, будет с благодарностью.