Ecto.Adapters.SQL.stream是否会发起多次查询？

核心工作原理 #

• 该API基于数据库游标（Cursor）实现流式查询，不会一次性加载全量查询结果到应用内存，完全适配你处理大量数据导出CSV的场景，不会出现OOM问题。
• 调用时不会立刻执行SQL，只会返回一个标准的Elixir Enumerable 流结构，只有当你对流进行枚举消费时，才会真正触发查询执行。
• 运行时会自动在数据库侧开启游标，按照你指定的max_rows参数值分批拉取数据，处理完当前批次再拉下一批，全程应用内存仅保留当前批次的少量数据。
• 流的生命周期绑定在数据库事务内，使用时必须将流的消费逻辑包裹在Repo.transaction中，事务结束后会自动关闭游标，不会产生资源残留。

适配你场景的示例代码 #

alias Ecto.Adapters.SQL

# 替换为你自己的复杂原生SQL
raw_sql = "SELECT field1, field2, field3 FROM your_big_table WHERE condition = ?"
# 对应SQL的参数列表，无参数则传空列表
params = ["your_condition_value"]

Repo.transaction(fn ->
  SQL.stream(Repo, raw_sql, params, max_rows: 1000)
  # 提取每一批的行数据
  |> Stream.flat_map(& &1.rows)
  # 转换为CSV支持的格式，可对接nimble_csv等CSV编码库
  |> Stream.map(fn [field1, field2, field3] -> [field1, to_string(field2), DateTime.to_iso8601(field3)] end)
  # 流式写入本地CSV文件
  |> Stream.into(File.stream!("export_data.csv"))
  |> Stream.run()
end)

# 后续执行CSV文件上传S3的逻辑即可

注意事项 #

• 如果你使用的是MySQL数据库，需要在Repo的连接配置中添加cursor: true参数，否则流式查询不会生效；PostgreSQL无需额外配置默认支持。
• max_rows参数可根据单条数据的大小调整，常规场景下1000~5000是比较合理的取值，取值过大容易提升内存占用，过小会增加和数据库的交互次数拉低效率。

学习资源推荐 #

• Ecto官方文档的SQL适配器章节，有流式查询的详细参数说明
• Elixir标准库Stream模块的官方文档，可熟悉各类流式操作的用法
• Elixir社区的Ecto实战教程，包含大量大结果集处理的落地案例

内容的提问来源于stack exchange，提问作者Kousik