Elixir:我可以使用 Stream.resource 逐步读取大型数据文件吗？

Question

我知道如何使用 Stream.resource() 来获取前 5 行
一个文件并将它们放在一个列表中。

str = Stream.resource(fn -> File.open!("./data/fidap011.mtx") end,
                fn file ->
                  case IO.read(file, :line) do
                    data when is_binary(data) -> {[data], file}
                    _ -> {:halt, file}
                  end
                end,
                fn file -> File.close(file) end)
str |>  Enum.take(5)

但是我如何从同一个流中获取接下来的 5 行呢？
如果我再次输入:

str |>  Enum.take(5)

我只是得到相同的前 5 行。

我在这里遗漏了一些明显的东西吗？

最终，我希望从我的流中读取足够的数据来生成一些进程
处理该数据。当其中一些过程完成时，我希望
从同一流中读取更多内容，从而处理下一组数据等。
Stream.chunk() 应该在这里发挥作用吗？
但是没有一个例子，我似乎无法直觉如何。

编辑 - 之后的几次设计迭代!

出于我的目的，不使用 Stream 更容易。
相反，我简单地使用创建一个文件指针/进程

{:ok, fp} = File.open("data/fidap011.mtx")

然后我实际上将该 fp 传递给 30000 个不同的衍生进程
他们随时都可以轻松阅读。
这些进程中的每一个都通过读取其新状态来更改其状态
文件中的变量。在下面的模块 oR和 vR是两个
接收消息的“路由器”进程 - 代码是稀疏的一部分
矩阵/向量乘数。

defmodule M_Cells do
 @moduledoc """
 Provides matrix related code
 Each cell process serves for that row & col
 """

 defp get_next_state( fp ) do
    case IO.read( fp, :line ) do
        data when is_binary(data) ->
            [rs,cs,vs] = String.split( data )
            r = String.to_integer(rs)
            c = String.to_integer(cs)
            v = String.to_float(vs)
            {r,c,v}
        _ -> 
            File.close( fp )
            :fail
    end
 end


 defp loop(fp, r,c,v, oR,vR) do
  # Maintains state of Matrix Cell, row, col, value 
  # receives msgs and responds
   receive do

    :start  ->  
        send vR, { :multiply, c, self() }  # get values for operands via router vR
        loop(fp, r,c,v, oR,vR)

    { :multiply, w } ->  # handle request to multiply by w and relay to router oR
        send oR, { :sum, r, v*w }
        case get_next_state( fp ) do # read line from file and fill in rcv
            {r1,c1,v1} ->
                send vR, { :multiply, c1, self() }
                loop(fp, r1,c1,v1, oR,vR)
            _ -> ## error or end of file etc
              ##IO.puts(":kill rcv: #{r},#{c},#{v}")
              Process.exit( self(), :kill )
        end
   end
 end

 # Launch each matrix cell using iteration by tail recursion
 def launch(_fp, _oR,_vR, result, 0) do
   result |> Enum.reverse # reverse is cosmetic, not substantive
 end

 def launch(fp, oR,vR, result, count) do
    #IO.inspect count
    case get_next_state( fp ) do
        {r,c,v} ->
            pid = spawn fn -> loop( fp, r,c,v, oR,vR) end
            launch( fp, oR,vR, [pid|result], count-1 )

        _ -> ## error or end of file etc, skip to count 0
            launch( fp, oR,vR, result, 0 )
    end
 end

end

请享用!

Answer 1

作为旁注，从文件创建流是一项常见任务。这已经处理好了，所以你可以简单地使用 File.stream!/3 创建流，无需使用 Stream.resource/3 直接地。

关于你原来的问题:是的，你是对的， Stream.chunk_every/2 是去这里的路。它会懒惰地将流分成提供大小的 block :

File.stream!("./data/fidap011.mtx") |> Stream.chunk_every(5) |> Enum.each(fn chunk ->
  # do something with chunk
end)