Clojure / Java: Die meisten effektive Methode zur Bandbreitenverbrauch zu minimieren, wenn sie auf einem Strom von Amazon S3 Daten komplexe Operationen

Question

Ich bin Streaming Durchführung liest ein Objekt BufferedReader verwenden.

Ich muss mit diesem Objekt zwei Dinge tun:

1. Direkt an einen SuperCSV csv Leser
2. Rufen Sie die Rohzeilen und halten sie in einer (Clojure) faul Sequenz

Derzeit bin ich mit zwei verschiedenen BufferedReaders verwenden: ein als Argument für eine SuperCSV CSV-Reader-Klasse und man die faulen Folge von Rohzeilen zu initialisieren. Ich bin zweimal das S3-Objekt effektiv herunterzuladen, was teuer ist ($) und langsam.

Einer meiner Kollegen weist darauf hin, dass etwas analog zu einem Unix „tee“ Befehl ist das, was ich suche. Ein BufferedReader, die irgendwie „split“ sein könnte, einen Teil der Daten herunterladen und eine Kopie an beide passieren die faule Sequenz und csv-Reader Funktionalität nützlich sein würde.

Ich untersuche derzeit auch, ob es möglich wäre, die faul Sequenz in einem BufferedReader wickeln und übergibt , die zu Super csv. Ich habe einige Java Heap-Speicher Probleme hatte, wenn sehr große faul Sequenzen an mehrere Verbraucher weiterzugeben, so dass ich Art von besorgt bin über diese Lösung verwendet wird.

Eine andere Lösung ist das Herunterladen nur die Datei lokal und dann zwei Ströme auf dieser Datei zu öffnen. Dadurch entfällt die ursprüngliche Motivation hinter Streaming. Ermöglicht die Verarbeitung der Datei so schnell zu beginnen, wie Daten beginnen ankommen

Die Endlösung, und eine, die ich nur andere funktioniert, wenn nichts betrachten würde, ist meine eigenen CSV-Leser, dass die Renditen sowohl geparsten CSV und die ursprüngliche unparsed Linie zu implementieren. Wenn Sie einen sehr soliden CSV-Reader verwendet haben, die sowohl einen Java-Hash der geparsten CSV-Daten und die ursprüngliche unparsed Linie zurückkehren können, lassen Sie es mich wissen!

Danke!

Answer 1

Die Lösung war einen einzigen BufferedReader für alle Zugriffe zu verwenden und dann reset () jedes Mal ing es in Funktionalität übergeben wird, dass der Bedarf von Anfang an lesen.

Answer 2

würde ich geneigt sein, eine seq von Linien aus dem Netz zu gehen, mit dem Erstellen, und geben Sie dann das zu über jedoch viele Prozesse müssen auf diesem seq zu arbeiten; persistente Datenstrukturen sind auf diese Weise kühlen. Im Fall des Müssens eine seq von Strings in einen Reader zu drehen, dass Sie die api SuperCSV Hand von können, dies scheint zu funktionieren:

(import '[java.io Reader StringReader])

(defn concat-reader
  "Returns a Reader that reads from a sequence of strings."
  [lines]
  (let [srs (atom (map #(StringReader. %) lines))]
    (proxy [Reader] []
      (read 
        ([] 
          (let [c (.read (first @srs))]
            (if (and (neg? c) (swap! srs next))
              (.read this)
              c)))
        ([cbuf] 
          (.read this cbuf 0 (count cbuf)))
        ([cbuf off len]
          (let [actual (.read (first @srs) cbuf off len)]
            (if (and (neg? actual) (swap! srs next))
              (.read this cbuf off len)
              actual))))
      (close [] ))))

z.

user=> (def r (concat-reader ["foo" "bar"]))
#'user/r
user=> (def cbuf (char-array 2))
#'user/cbuf
user=> (.read r cbuf)
2
user=> (seq cbuf)
(\f \o)
user=> (char (.read r))
\o
user=> (char (.read r))
\b