Wiedergabe 2 Scala - Der beste Weg, eine große CSV-Datei mit Iteratee hochladen, um jede Zeile reaktiv

Question

Ich möchte zu verarbeiten Play2 verwenden, um eine sehr große CSV-Dateien (milions von Linien) auf Elasticsearch hochladen. Ich habe den folgenden Code geschrieben, der einwandfrei funktioniert.

Ich bin nicht zufrieden mit der Art, wie ich die HTTP-Antwort-Header in den ersten Brocken überspringe. Es sollte einen Weg geben, diesen iteratee mit einem ersten Iteratee zu verketten, der den http-Header überspringt und direkt in den Done-Zustand wechselt, aber ich habe nicht gefunden, wie.

Wenn jemand

object ReactiveFileUpload extends Controller { 
    def upload = Action(BodyParser(rh => new CsvIteratee(isFirst = true))) { 
    request => 
     Ok("File Processed") 
    } 
} 

case class CsvIteratee(state: Symbol = 'Cont, input: Input[Array[Byte]] = Empty, lastChunk: String = "", isFirst: Boolean = false) extends Iteratee[Array[Byte], Either[Result, String]] { 
    def fold[B](
       done: (Either[Result, String], Input[Array[Byte]]) => Promise[B], 
       cont: (Input[Array[Byte]] => Iteratee[Array[Byte], Either[Result, String]]) => Promise[B], 
       error: (String, Input[Array[Byte]]) => Promise[B] 
       ): Promise[B] = state match { 
    case 'Done => 
     done(Right(lastChunk), Input.Empty) 

    case 'Cont => cont(in => in match { 
     case in: El[Array[Byte]] => { 
     // Retrieve the part that has not been processed in the previous chunk and copy it in front of the current chunk 
     val content = lastChunk + new String(in.e) 
     val csvBody = 
      if (isFirst) 
      // Skip http header if it is the first chunk 
      content.drop(content.indexOf("\r\n\r\n") + 4) 
      else content 
     val csv = new CSVReader(new StringReader(csvBody), ';') 
     val lines = csv.readAll 
     // Process all lines excepted the last one since it is cut by the chunk 
     for (line <- lines.init) 
      processLine(line) 
     // Put forward the part that has not been processed 
     val last = lines.last.toList.mkString(";") 
     copy(input = in, lastChunk = last, isFirst = false) 
     } 
     case Empty => copy(input = in, isFirst = false) 
     case EOF => copy(state = 'Done, input = in, isFirst = false) 
     case _ => copy(state = 'Error, input = in, isFirst = false) 
    }) 

    case _ => 
     error("Unexpected state", input) 

    } 

    def processLine(line: Array[String]) = WS.url("http://localhost:9200/affa/na/").post(
    toJson(
     Map(
     "date" -> toJson(line(0)), 
     "trig" -> toJson(line(1)), 
     "code" -> toJson(line(2)), 
     "nbjours" -> toJson(line(3).toDouble) 
    ) 
    ) 
) 
} 

Answer 1

Zur Kette zwei iteratees zusammen helfen können, verwenden flatMap.

val combinedIteratee = firstIteratee.flatMap(firstResult => secondIteratee) 

Oder ein für das Verständnis mit:

val combinedIteratee = for { 
    firstResult <- firstIteratee 
    secondResult <- secondIteratee 
} yield secondResult 

Sie können flatMap verwenden so viele iteratees zusammen sequenzieren, wie Sie möchten.

Sie könnte so etwas wie zu tun:

val headerAndCsvIteratee = for { 
    headerResult <- headerIteratee 
    csvResult <- csvIteratee 
} yield csvResult