エリクシルのCSVストリームへのストリーム

Question

ファイルの最初の行がヘッダーである場合、大きな大文字をcsvのデータとして解析する必要があります。ライブラリ:csvは既に私にストリームのリストを与えています。最初の行から構造を推測する必要がありますが、それを無視して、指定された構造のマップのストリームを生成する必要があります。

私はこれが好き：

data.csv

a,b 
1,2 
3,4 
... 

CSV.stream_map（ファイル名）出力

{a: 1, b: 2} #1 
{a: 3, b: 4} #2 
... 

私はStream.transformに探したがわからなかったました最初の要素をスキップする方法を説明します。構造はアキュムレータに格納することができます。

Answer 1

私が見つけたようにcsvモジュールがすでにこれを行っていますが、私はこれを自分で実装する方法も見つけました。それはあなたがStream.transformコールバックに空のリスト[]を送り返した場合、どの要素がストリームに押し込まれていないされることを判明：

def map_stream(enum) do 
    enum 
    |> Stream.transform(:first, &structure_from_header/2) 
end 

#The accumulator starts as :first, the its the structure of the csv 
#that is the first line 
def structure_from_header(line, :first), 
    do: { [ ], line } #<=================== Here is the trick 

def structure_from_header(line, structure) do 
    map = 
     structure 
     |> Enum.zip(line) 
     |> Enum.into(%{}) 

{ [ map ], structure } 
end 

Answer 2

（docsに述べたように）あなたがCSV.decode/2に2番目の引数としてheaders: trueを渡すと、自動的に最初の行をキー名として使用し、次のすべての行についてMapを返します。

iex(1)> CSV.decode(File.stream!("data.csv"), headers: true) |> Enum.to_list 
[%{"a" => "1", "b" => "2"}, %{"a" => "3", "b" => "4"}] 

data.csv含まれています

a,b 
1,2 
3,4 

Answer 3

私が考える2つのオプションがあります。ここでは、チャンクサイズを設定して、ファイル全体をメモリにロードせず、そのセットで作業することができます。データを解析する必要がある場合は、フローソリューションを使用しないでください。私は両方ともヘッダーをスキップする方法を示します。マップ構造の作成に関しては、structsを見て、構造を利用してマップセットの構造を作成することができます。あなたがたくさんの列を持っているなら私は地図ではなく地図を提案します。

def stream_parse(file_path, chunk_size) do 
    file_path 
    |> File.stream! 
    |> Stream.drop(1) 
    |> Stream.map(&String.split(&1, ",")) 
    |> Stream.chunk(chunk_size, chunk_size, []) 
    |> Stream.map(&MapSet.new(&1)) 
end 

def flow_parse(file_path, chunk_size) do 
    file_path 
    |> File.stream!(read_ahead: chunk_size) 
    |> Stream.drop(1) 
    |> Flow.from_enumerable 
    |> Flow.map(&String.split(&1, ",")) 
    |> Flow.partition 
    |> Flow.map(&MapSet.new(&1) 
end