3、数据源接入（一）：文件数据源（CSV、JSON、Parquet）的读取与写入

数据流水线的第一关，就是把数据从源文件里「请」进来。

我个人习惯把文件数据源分成三类：文本型（CSV、JSON）、列式存储型（Parquet），以及后面会讲到的二进制/日志型。这一章我们先搞定前两种。

你想想看，如果连文件都读不对，后面再花哨的ETL都是白搭。所以这一节，咱们把Pandas和PyArrow这两个工具用熟，把CSV、JSON、Parquet的读写细节吃透。

3.1 先搞清楚：Pandas 和 PyArrow 到底怎么选？

很多新手上来就问：「我用Pandas读CSV，够用吗？」

我的回答是：够用，但不一定好用。

Pandas的read_csv()确实方便，但它在底层用的是Python原生的CSV解析器，遇到大文件或者复杂数据类型（比如时间戳带时区、嵌套JSON）就会很吃力。我在项目中遇到过几次，用Pandas读一个5GB的CSV，内存直接爆了——后来换成PyArrow的read_csv()，内存占用降了40%。

PyArrow是Apache Arrow的Python绑定，它的核心优势是零拷贝、列式内存布局。说白了，它读文件的时候不会把数据来回倒腾，而是直接映射到内存里，所以速度更快、内存更省。

我建议这样选：

场景	推荐工具	原因
小文件（< 100MB），快速验证	Pandas	API简单，生态成熟
大文件（> 1GB），生产环境	PyArrow	内存效率高，支持并行读取
需要与NumPy/SciPy交互	Pandas	DataFrame直接转NumPy数组
需要与Spark/Flink集成	PyArrow	Arrow是大数据生态的通用内存格式

3.2 CSV：最通用的格式，但坑最多

CSV看起来简单，但实际项目中踩过的坑能写满一页纸。咱们一个一个说。

3.2.1 用Pandas读CSV

基础用法我就不啰嗦了，直接上代码：

import pandas as pd

# 最基础的读取
df = pd.read_csv('data.csv')

# 指定分隔符（比如制表符）
df = pd.read_csv('data.tsv', sep='\t')

# 只读前100行，适合预览
df_sample = pd.read_csv('data.csv', nrows=100)

# 指定列的数据类型，避免Pandas猜错
dtype_dict = {
    'user_id': 'int32',
    'age': 'int8',
    'salary': 'float32'
}
df = pd.read_csv('data.csv', dtype=dtype_dict)

嗯，这里要注意：一定要指定dtype。Pandas默认会自己推断类型，但推断过程很慢，而且容易出错。比如一个全是数字的列，如果中间混了一个空值，Pandas可能就把它当成float了。

3.2.2 用PyArrow读CSV

PyArrow的API稍微不一样，但更严谨：

import pyarrow.csv as pv
import pyarrow as pa

# 读取CSV
table = pv.read_csv('data.csv')

# 指定解析选项
parse_options = pv.ParseOptions(
    delimiter=',',
    quote_char='"',
    double_quote=True,
    escape_char=None
)

# 指定转换选项（类似dtype）
convert_options = pv.ConvertOptions(
    column_types={
        'user_id': pa.int32(),
        'age': pa.int8(),
        'salary': pa.float32()
    }
)

table = pv.read_csv('data.csv',
                    parse_options=parse_options,
                    convert_options=convert_options)

# 转成Pandas DataFrame
df = table.to_pandas()

3.2.3 CSV写入

写入相对简单，但有几个参数值得注意：

# Pandas写入
df.to_csv('output.csv', index=False, encoding='utf-8-sig')

# PyArrow写入
import pyarrow.csv as pv
pv.write_csv(table, 'output.csv')

# 带选项的写入
pv.write_csv(table, 'output.csv',
             write_options=pv.WriteOptions(
                 delimiter=',',
                 include_header=True,
                 quoting_style='needed'
             ))

我个人习惯用utf-8-sig编码，这样Excel打开不会乱码。你想想看，数据工程师交付的文件，业务方用Excel打开全是乱码，那场面多尴尬。

3.3 JSON：灵活但解析成本高

JSON在API数据交换中很常见，但作为数据源，它有几个「天生缺陷」：

• 嵌套结构：JSON天然支持嵌套，但扁平化需要额外处理
• 类型松散：同一个字段在不同记录里可能是不同类型
• 解析慢：JSON解析比CSV慢3-5倍

3.3.1 用Pandas读JSON

import pandas as pd

# 读取标准JSON（每行一个对象）
df = pd.read_json('data.json', lines=True)

# 读取嵌套JSON，自动展开
df = pd.read_json('nested_data.json')

# 指定方向
df = pd.read_json('data.json', orient='records')

这里有个坑：lines=True表示每行是一个独立的JSON对象，这种格式叫JSON Lines（.jsonl）。很多日志文件就是这种格式，读取效率比标准JSON高得多。

3.3.2 用PyArrow读JSON

import pyarrow.json as pj

# 读取JSON
table = pj.read_json('data.json')

# 读取JSON Lines
table = pj.read_json('data.jsonl', format='line-delimited')

# 转成Pandas
df = table.to_pandas()

3.3.3 JSON写入

# Pandas写入
df.to_json('output.json', orient='records', lines=True)

# PyArrow写入
import pyarrow.json as pj
pj.write_json(table, 'output.json')

写入时我一般用orient='records'，这样每行一个JSON对象，方便后续按行处理。

3.4 Parquet：列式存储的王者

Parquet是我在生产环境中最常用的格式。为什么？

• 压缩率高：列式存储天然适合压缩，比CSV小3-5倍
• 读取快：只读需要的列，IO开销小
• Schema自带：数据类型、字段名都写在文件里，不会猜错

3.4.1 用Pandas读写Parquet

import pandas as pd

# 读取Parquet
df = pd.read_parquet('data.parquet')

# 只读指定列
df = pd.read_parquet('data.parquet', columns=['user_id', 'age'])

# 写入Parquet
df.to_parquet('output.parquet', compression='snappy')

注意：Pandas的read_parquet()底层依赖PyArrow或fastparquet。我建议统一用PyArrow引擎：

df = pd.read_parquet('data.parquet', engine='pyarrow')
df.to_parquet('output.parquet', engine='pyarrow', compression='zstd')

3.4.2 用PyArrow读写Parquet

import pyarrow.parquet as pq

# 读取Parquet
table = pq.read_table('data.parquet')

# 只读指定列
table = pq.read_table('data.parquet', columns=['user_id', 'age'])

# 读取元数据（不加载数据）
metadata = pq.read_metadata('data.parquet')
print(metadata.num_rows)  # 行数
print(metadata.num_columns)  # 列数

# 写入Parquet
pq.write_table(table, 'output.parquet',
               compression='zstd',
               row_group_size=100000)

3.5 三种格式对比

为了让你看得更清楚，我整理了一张对比表：

特性	CSV	JSON	Parquet
可读性	高（文本编辑器可看）	中（嵌套时难读）	低（二进制）
压缩率	低（1x）	中（1.5x）	高（3-5x）
读取速度	慢（全量扫描）	慢（解析开销大）	快（列裁剪+谓词下推）
Schema支持	无（需要手动指定）	弱（类型松散）	强（自带Schema）
适用场景	小文件、临时数据	API数据、日志	生产环境、大数据

3.6 本章知识体系

下面这张图帮你理清三种格式的读写路径和工具选择：

3.7 避坑指南

最后，分享几个我实际项目中踩过的坑：

• 编码问题：CSV文件可能是GBK、UTF-8、UTF-8 with BOM。我建议统一用encoding='utf-8-sig'，兼容性最好。
• 空值处理：CSV里的空单元格，Pandas默认读成NaN，PyArrow读成null。如果你后续要写回数据库，注意类型转换。
• 大JSON文件：不要用pd.read_json()读大JSON，内存会炸。改用pd.read_json(lines=True)或者PyArrow。
• Parquet分区：如果Parquet文件是按日期分区的，用pq.read_table()配合filters参数，只读需要的分区，速度能快10倍。