Elixir Datei Stromwiederverwendung/Gabel

Question

Ich habe eine extrem große CSV-Datei mit Daten und ganzen Zahlen. Ich muss für jeden Dateieintrag einen Ecto-Datensatz erstellen. Das Problem ist, dass die Ganzzahlen basierend auf dem minimalen Zeitunterschied zwischen den Daten benachbarter Datensätze manipuliert werden müssen. Ich habe versucht, den Strom bis zu einem Punkt zu verarbeiten, an eine Variable zu binden und sie dann für zwei verschiedene Berechnungen zu verwenden. Die zweite Berechnung erhält jedoch einen leeren Stream. Es scheint, dass jeder Zugriff auf den Stream, was auch immer ich habe entfernt. Gibt es eine Möglichkeit, einen Stream erneut zu verwenden/fork/clone/dup/something? Ich kenne RX Streams hat dieses Konzept. Ich versuchte über einen Weg nachzudenken, dies in einer einzigen Kette zu tun, aber ich kam leer aus. Hier ist im Grunde der Fluss Ich habe versucht:

def do_something(path) do 
    {:ok, file} = File.open(path) 
    stream = file 
    |> IO.stream(:line) 
    |> Stream.map(&String.split(&1, ",")) 

    dates = stream_to_dates(stream) # stream 
    factor = dates_to_factor(dates) # float 
    values = stream_to_values(stream, factor) # stream 

    Stream.zip(dates, values) 
end 

Ich bin in der Lage, die Daten zu berechnen und dann dem Faktor, aber gleich danach der Strom und Daten sind beide leer Ströme, also die Werte leer ist, so der Reißverschluss ist leer ...

Answer 1

Sie Stream.transform verwenden können, um einen Strom von Paaren aus einem gegebenen Strom zu erzeugen:

def pairs(stream) do 
    Stream.transform(stream, nil, fn(x, last) -> 
    # The first element is the list of values to return at this point, 
    # the second one is the new accumulator 
    {[{last, x}], x} 
    end) 
    # Drop the first pair of {nil, something} 
    |> Stream.drop(1) 
end 

Dann:

iex(1)> 1..1000 |> Stream2.pairs |> Enum.take(5) 
[{1, 2}, {2, 3}, {3, 4}, {4, 5}, {5, 6}] 

Sie sollten dann in der Lage sein, ein Paar nach |> Stream.map(&String.split(&1, ",")) zu verwenden, um Paare benachbarter Datensätze zu erhalten. Wenn Sie größere Stücke benötigen, kann die Funktion verallgemeinert werden.