Elixir/Ecto如何对关联表数据进行分批预加载

Ecto 原生 Ecto.preload/3 没有直接内置 batch_size 配置项，但可以通过以下几种方式实现分批预加载的需求，避免单次查询拉取超大量评论数据超时：

方案1：手动拆分ID分批查询后关联 #

先拉取所有符合条件的Post，再将Post ID拆分为固定批次分别查询对应评论，最后手动关联到Post对象，实现成本最低：

# 拉取所有需要处理的Post
posts = Post.filter(needs_review: true) |> Repo.all()
post_ids = Enum.map(posts, & &1.id)

# 按每1000个Post为一批拆分查询评论
comments = 
  post_ids
  |> Enum.chunk_every(1000)
  |> Enum.flat_map(fn batch_ids ->
    from(c in Comment, where: c.post_id in ^batch_ids) |> Repo.all()
  end)

# 按post_id分组后关联到对应Post
comments_by_post = Enum.group_by(comments, & &1.post_id)
posts_with_comments = Enum.map(posts, fn post ->
  %{post | comments: Map.get(comments_by_post, post.id, [])}
end)

方案2：流式分批处理（适合超大数据量） #

如果连Post总量都非常大，不适合一次性全部加载到内存，可以用Repo.stream边拉边处理，内存占用会非常低：

Repo.transaction(fn ->
  Post.filter(needs_review: true)
  |> Repo.stream(chunk_size: 1000)
  |> Stream.map(fn post ->
    # 逐批关联评论
    %{post | comments: Repo.all(from c in Comment, where: c.post_id == ^post.id)}
  end)
  |> Stream.each(fn post_with_comments ->
    # 你的清理逻辑写在这里
  end)
  |> Stream.run()
end)

方案3：使用第三方库实现类预期语法 #

如果希望使用和你期望接近的语法，可以引入ecto_preloader工具库，直接支持分批预加载配置：

1. 先在mix.exs中添加依赖

defp deps do
  [
    # 其他依赖
    {:ecto_preloader, "~> 0.2"}
  ]
end

2. 调用方式和你预期的逻辑基本一致：

Post.filter(needs_review: true)
|> Repo.all()
|> EctoPreloader.preload([:comments], batch_size: 1000)

优化建议 #

如果你的清理任务不需要把全量Post和Comment数据加载到应用层做复杂逻辑，更推荐直接用数据库关联操作完成批量处理，性能可以提升几个量级：

# 示例：直接批量删除需审核Post对应的所有评论，不需要拉取数据到应用层
from(c in Comment,
  join: p in Post,
  on: c.post_id == p.id,
  where: p.needs_review == true
)
|> Repo.delete_all()

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