1. 引言：多文件数据合并与挑战
在数据分析和处理中，我们经常会遇到需要处理大量结构相同但分散在多个文件中的数据。例如，一系列按产品或日期划分的CSV文件：data_product_1.csv、data_product_2.csv等。Polars作为一款高性能的数据框库，提供了便捷的方式来合并这些文件，例如使用通配符直接读取：pl.read_csv("data_*.csv")。
然而，一个常见的需求是在合并后的数据中保留每条记录的原始文件信息。例如，将文件名中的“product_1”提取出来作为新的“product_code”列。虽然可以通过逐个文件加载、添加列、然后合并的传统方法实现，但这可能无法充分利用Polars在处理大量数据时的性能优势，尤其是在面对大数据集时。本教程将深入探讨如何利用Polars的惰性计算特性，以一种高效且并行的方式解决这一问题。
2. 传统方法与Polars的惰性优势
对于多文件处理并添加源信息的需求，一种直观但可能效率不高的方法是：
- 遍历所有目标文件。
- 对每个文件，使用 pl.read_csv() 加载数据。
- 在加载后的DataFrame中添加一列，包含该文件的标识信息（例如文件名）。
- 将所有处理过的DataFrame收集到一个列表中。
- 最后使用 pl.concat() 将它们合并。
这种方法在文件数量不多或文件较小时尚可接受，但当文件数量庞大或单个文件体积较大时，会因为频繁的I/O操作和内存占用而导致性能瓶颈。此外，某些数据库系统（如DuckDB）提供了在读取CSV时直接添加文件名列的功能（例如 read_csv_auto('data_*.csv', filename = true)），这显示了此类功能的实用性。
Polars虽然在 read_csv 或 scan_csv 中尚未直接内置 filename=true 这样的参数（截至本文撰写时，此功能仍在社区讨论中），但其强大的惰性计算（LazyFrames）机制为我们提供了一个优雅且高性能的解决方案。通过利用惰性操作，Polars可以构建一个执行计划，在实际执行前进行优化，并能以并行方式处理多个文件，从而显著提升效率。
3. 利用Polars LazyFrames高效处理多文件
Polars的惰性API是解决此问题的核心。它允许我们定义一系列数据转换操作，而无需立即加载或计算数据。只有当明确调用 .collect() 方法时，Polars才会执行这些操作并返回一个具体的DataFrame。
3.1 惰性扫描文件 (scan_csv)
与 pl.read_csv() 直接加载数据不同，pl.scan_csv() 返回一个 LazyFrame 对象。LazyFrame 只是一个操作计划的表示，它不会立即读取文件内容，从而节省了内存和计算资源。
import polars as pl
from pathlib import Path
# 假设当前目录下有 data_product_1.csv, data_product_2.csv 等文件
# 为了演示，我们先创建一些模拟文件
file_contents = """data,value
2000-01-01,1
2000-01-02,2
"""
Path("data_product_1.csv").write_text(file_contents)
file_contents_2 = """data,value
2000-01-01,3
2000-01-02,4
"""
Path("data_product_2.csv").write_text(file_contents_2)
file_contents_3 = """data,value
2000-01-01,4
2000-01-02,5
"""
Path("data_product_3.csv").write_text(file_contents_3)
# 遍历所有匹配的文件，并为每个文件创建一个LazyFrame
csv_lazyframes = []
for f_path in Path().glob("data_*.csv"):
# 使用 scan_csv 惰性读取文件
lazy_df = pl.scan_csv(f_path)
csv_lazyframes.append(lazy_df)
# 此时，数据尚未被实际读取
print(f"创建了 {len(csv_lazyframes)} 个 LazyFrame 对象。")3.2 添加源信息列 (with_columns)
在创建 LazyFrame 后，我们可以立即在其上链式调用各种转换操作，例如添加新列。这里，我们将利用 f_path.name 获取文件名，并将其作为新的 product_code 列添加到每个 LazyFrame 中。pl.lit() 用于将Python字符串转换为Polars字面量表达式。
import polars as pl
from pathlib import Path
# (省略模拟文件创建部分，假设文件已存在)
# 遍历所有匹配的文件，并为每个文件创建一个LazyFrame，同时添加product_code列
csv_lazyframes_with_product_code = [
pl.scan_csv(f_path).with_columns(product_code=pl.lit(f_path.name))
for f_path in Path().glob("data_*.csv")
]
# 此时，每个LazyFrame都包含一个添加product_code列的指令，但数据仍未加载
print(f"创建了 {len(csv_lazyframes_with_product_code)} 个包含 'product_code' 列指令的 LazyFrame 对象。")3.3 并行合并 (concat) 与数据收集 (collect)
Polars的 pl.concat() 函数不仅可以合并Eager DataFrame，也能高效地合并 LazyFrame 列表。当合并 LazyFrame 时，pl.concat() 默认会利用多核CPU并行处理各个文件的读取和转换操作，从而大大加快处理速度。最后，调用 .collect() 方法会触发所有惰性操作的实际执行，将结果物化为一个最终的Polars DataFrame。
import polars as pl
from pathlib import Path
# 1. 创建模拟数据文件 (如果尚未创建)
file_contents_1 = """data,value
2000-01-01,1
2000-01-02,2
"""
Path("data_product_1.csv").write_text(file_contents_1)
file_contents_2 = """data,value
2000-01-01,3
2000-01-02,4
"""
Path("data_product_2.csv").write_text(file_contents_2)
file_contents_3 = """data,value
2000-01-01,4
2000-01-02,5
"""
Path("data_product_3.csv").write_text(file_contents_3)
# 2. 核心解决方案：使用LazyFrames处理和合并文件
# 遍历文件，创建LazyFrame，并添加文件名作为product_code列
lazy_frames = [
pl.scan_csv(f_path).with_columns(product_code=pl.lit(f_path.name))
for f_path in Path().glob("data_*.csv")
]
# 使用pl.concat合并所有LazyFrames，并调用.collect()执行计算
# pl.concat在处理LazyFrames时会默认尝试并行化读取和转换操作
final_df = pl.concat(lazy_frames).collect()
# 3. 打印结果
print("最终合并的DataFrame:")
print(final_df)
# 清理模拟文件
Path("data_product_1.csv").unlink()
Path("data_product_2.csv").unlink()
Path("data_product_3.csv").unlink()输出示例:
最终合并的DataFrame: shape: (6, 3) ┌────────────┬───────┬────────────────────┐ │ data ┆ value ┆ product_code │ │ --- ┆ --- ┆ --- │ │ str ┆ i64 ┆ str │ ╞════════════╪═══════╪════════════════════╡ │ 2000-01-01 ┆ 1 ┆ data_product_1.csv │ │ 2000-01-02 ┆ 2 ┆ data_product_1.csv │ │ 2000-01-01 ┆ 3 ┆ data_product_2.csv │ │ 2000-01-02 ┆ 4 ┆ data_product_2.csv │ │ 2000-01-01 ┆ 4 ┆ data_product_3.csv │ │ 2000-01-02 ┆ 5 ┆ data_product_3.csv │ └────────────┴───────┴────────────────────┘
4. 核心优势与注意事项
- 惰性计算与性能优化： 这种方法的核心优势在于惰性计算。Polars可以构建一个全局的执行计划，并对所有操作进行优化，避免不必要的中间数据加载和存储。对于大量文件或大型文件，这能显著减少内存占用和提高处理速度。
- 并行处理： pl.concat 在处理 LazyFrame 列表时，会默认利用多核CPU并行读取和处理各个文件，进一步加速数据整合过程。
- 灵活性： 在 with_columns 步骤中，你可以执行更复杂的逻辑来从文件名中提取信息。例如，如果文件名是 data_product_1.csv，你可以使用字符串操作 pl.col("product_code").str.extract(r"product_(\d+)") 来提取纯粹的产品编号。
- 文件命名规范： 从文件名提取信息的前提是文件命名具有一致性和可解析性。清晰的命名约定将使信息提取变得简单可靠。
- Polars的演进： 尽管目前需要手动添加文件名列，但Polars社区正在积极开发新功能。未来版本中可能会直接在 read_csv 或 scan_csv 中提供类似 filename=true 的参数，届时处理方式可能会更加简化。
5. 总结
通过本教程，我们学习了如何利用Polars的 scan_csv、with_columns 和 concat 结合 LazyFrame 的特性，高效地处理多个CSV文件，并在合并过程中为每条记录添加源文件信息。这种方法不仅解决了特定场景下的数据处理需求，更展示了Polars在处理大数据集时卓越的性能和灵活性。掌握Polars的惰性API是提升数据处理效率的关键，尤其适用于需要复杂转换和大规模数据整合的场景。
