用 https://github.com/flow-php/etl，基于 Generator，大文件也不会撑爆内存。

  composer require flow-php/flow flow-php/etl-adapter-csv flow-php/etl-adapter-doctrine

  场景：CSV → 清洗 → 写入数据库

  <?php
  require 'vendor/autoload.php';

  use function Flow\ETL\Adapter\CSV\from_csv;
  use function Flow\ETL\Adapter\Doctrine\to_dbal_table;
  use Flow\ETL\Flow;
  use Flow\ETL\Row;
  use Doctrine\DBAL\DriverManager;

  $db = DriverManager::getConnection(['url' => 'mysql://root:pass@localhost/warehouse']);

  (new Flow())
      // E：从 CSV 抽数据（逐行读，不会一次性全加载进内存）
      ->read(from_csv('orders.csv'))

      // T：清洗和转换
      ->filter(fn(Row $r) => $r->valueOf('amount') > 0)           // 过滤无效行
      ->map(fn(Row $r) => $r
          ->rename('order_id', 'id')                              // 改列名
          ->set('amount',  (float) $r->valueOf('amount'))         // 类型转换
          ->set('created', new \DateTimeImmutable($r->valueOf('created_at'))) // 格式化日期
          ->remove('raw_notes')                                   // 删掉不要的列
      )

      // L：写入数据仓库
      ->write(to_dbal_table($db, 'dw_orders'))

      ->run();

  echo "done\n";

  多源合并（两张表 JOIN 后入库）

  use function Flow\ETL\DSL\{from_array, join};

  $orders    = (new Flow())->read(from_csv('orders.csv'));
  $customers = (new Flow())->read(from_csv('customers.csv'));

  (new Flow())
      ->read(from_array([]))          // 起点
      ->join($orders,    join()->left('customer_id'))
      ->join($customers, join()->left('id'))
      ->write(to_dbal_table($db, 'dw_order_detail'))
      ->run();

  大批量性能优化

  (new Flow())
      ->read(from_csv('big_file.csv'))
      ->map(fn(Row $r) => /* transform */)
      ->batchSize(500)                // 每500行批量写一次，减少 DB 往返
      ->write(to_dbal_table($db, 'dw_orders'))
      ->run();

  监控进度

  (new Flow())
      ->read(from_csv('orders.csv'))
      ->map(fn(Row $r) => $r)
      ->each(function (Row $r) {
          static $i = 0;
          if (++$i % 1000 === 0) echo "已处理 {$i} 行\n";
      })
      ->write(to_dbal_table($db, 'dw_orders'))
      ->run();

  ---
  核心就三步：read() 抽、map()/filter() 转、write() 载。Generator 架构保证不管文件多大内存都稳定，batchSize()
  控制写库频率。

一句话终极结论
 
flow-php/etl 是 PHP 的轻量化ETL工具，靠 Generator（生成器）逐行读取大文件，不会一次性把整个文件塞进内存，所以多大的CSV都不会撑爆内存；还能轻松做数据清洗、多文件关联、批量入库，三步搞定数据搬运。
 
下面全程大白话，把原理、代码、用法拆到完全易懂。
 
 
 
一、核心原理：为啥大文件不爆内存？
 
普通读文件是一次性把整个文件加载进内存，文件10GB内存就直接炸。
这个库用了 Generator（生成器）：
 
- 读文件时一行一行读，读完一行处理一行，处理完就扔掉
- 内存里永远只存当前一行数据，内存占用始终稳稳的
- 不管文件是100MB还是10GB，内存占用几乎不变
 
 
 
二、基础场景：CSV → 清洗 → 入库（ETL核心三步）
 
ETL 就是数据的抽 → 转 → 载，代码完全对应这三步：
 
1. read()：抽取数据（读CSV）
 
php
  
->read(from_csv('orders.csv'))
 
 
大白话：
从订单CSV文件里逐行读数据，不加载全文件，内存无压力。
 
2. map/filter：转换清洗（数据加工）
 
php
  
->filter(fn(Row $r) => $r->valueOf('amount') > 0) // 过滤金额≤0的无效行
->map(fn(Row $r) => $r
    ->rename('order_id', 'id') // 列名order_id改成id
    ->set('amount', (float)$r->valueOf('amount')) // 金额转成数字类型
    ->set('created', new \DateTimeImmutable($r->valueOf('created_at'))) // 日期格式化
    ->remove('raw_notes') // 删掉没用的列
)
 
 
大白话：
对每一行数据修修改改：
 
- 扔掉垃圾数据
- 改列名、转数据类型
- 删掉不需要的字段
- 格式化时间
 
3. write()：加载入库（写数据库）
 
php
  
->write(to_dbal_table($db, 'dw_orders'))
 
 
大白话：
把清洗好的数据，写入数据库的  dw_orders  表。
 
 
 
三、多源合并：两个CSV关联（JOIN）
 
php
  
$orders = (new Flow())->read(from_csv('orders.csv')); // 订单表
$customers = (new Flow())->read(from_csv('customers.csv')); // 客户表

(new Flow())
    ->read(from_array([]))
    ->join($orders, join()->left('customer_id')) // 按customer_id左关联订单
    ->join($customers, join()->left('id')) // 按id左关联客户
    ->write(to_dbal_table($db, 'dw_order_detail'))
    ->run();
 
 
大白话：
像数据库连表查询一样，把订单CSV和客户CSV按关联字段拼在一起，合并后再入库。
 
 
 
四、大批量性能优化
 
php
  
->batchSize(500) // 每500行批量写一次
 
 
大白话：
不每一行都插一次数据库（太频繁、太慢），攒够500行再批量插入，减少数据库往返次数，速度快几十倍。
 
 
 
五、进度监控：看处理了多少行
 
php
  
->each(function (Row $r) {
    static $i = 0;
    if (++$i % 1000 === 0) echo "已处理 {$i} 行\n";
})
 
 
大白话：
每处理1000行就打印一次进度，实时看处理到哪了，不会黑屏干等。
 
 
 
六、终极极简总结
 
1. 核心优势：Generator逐行读大文件，内存不爆炸
2. 核心逻辑：read读数据 → map/filter洗数据 → write写数据库
3. 进阶用法：多CSV关联合并、批量入库提速、实时监控进度
4. 适用场景：超大CSV清洗入库、数据同步、数据仓库导入