读取和写入 CSV 文件

Arrow 支持从 CSV 文件读取列式数据或将列式数据写入 CSV 文件。目前提供的功能如下：

• 多线程或单线程读取

• 输入文件的自动解压缩（基于文件扩展名，例如 my_data.csv.gz）

• 从 CSV 文件的第一行提取列名

• 基于列的类型推断，并可转换为 null、int64、float64、date32、time32[s]、timestamp[s]、timestamp[ns]、duration（来自数值字符串）、string 或 binary 数据

• 对 string 和 binary 列进行机会性字典编码（默认禁用）

• 检测各种 null 值的写法，如 NaN 或 #N/A

• 写入 CSV 文件，并提供配置精确输出格式的选项

用法

CSV 读取和写入功能可通过 pyarrow.csv 模块获得。在许多情况下，您只需调用 read_csv() 函数并传入您想要读取的文件路径即可

>>> from pyarrow import csv
>>> import pyarrow as pa
>>> import pandas as pd
>>> fn = 'tips.csv.gz'
>>> table = csv.read_csv(fn)
>>> table
pyarrow.Table
total_bill: double
tip: double
sex: string
smoker: string
day: string
time: string
size: int64
>>> len(table)
244
>>> df = table.to_pandas()
>>> df.head()
   total_bill   tip     sex smoker  day    time  size
0       16.99  1.01  Female     No  Sun  Dinner     2
1       10.34  1.66    Male     No  Sun  Dinner     3
2       21.01  3.50    Male     No  Sun  Dinner     3
3       23.68  3.31    Male     No  Sun  Dinner     2
4       24.59  3.61  Female     No  Sun  Dinner     4

要写入 CSV 文件，只需使用 pyarrow.RecordBatch 或 pyarrow.Table 以及路径或类文件对象调用 write_csv() 即可

>>> table = pa.table({'col1': [1, 2, 3], 'col2': ['a', 'b', 'c']})
>>> csv.write_csv(table, "tips.csv")
>>> with pa.CompressedOutputStream("tips.csv.gz", "gzip") as out:
...     csv.write_csv(table, out)

注意

写入器尚不支持所有 Arrow 类型。

自定义解析

如果需要读取结构不寻常的 CSV 文件并更改默认解析设置，您应该创建一个 ParseOptions 实例，并将其传递给 read_csv()

>>> def skip_handler(row):
...     pass
>>> table = csv.read_csv('tips.csv.gz', parse_options=csv.ParseOptions(
...    delimiter=";",
...    invalid_row_handler=skip_handler
... ))
>>> table
pyarrow.Table
col1,"col2": string
----
col1,"col2": [["1,"a"","2,"b"","3,"c""]]

可用的解析选项包括：

delimiter	CSV 数据中用于分隔各个单元格的字符。
quote_char	用于引用 CSV 值的字符（如果为 False，则不允许引用）。
double_quote	在带引号的 CSV 值中，两个引号是否表示数据中的单个引号。
escape_char	可选的用于转义特殊字符的字符（如果为 False，则不允许转义）。
newlines_in_values	是否允许在 CSV 值中使用换行符。
ignore_empty_lines	是否忽略 CSV 输入中的空行。
invalid_row_handler	无效行的可选处理程序。

另请参阅

有关更多示例，请参阅 ParseOptions。

自定义转换

要更改将 CSV 数据转换为 Arrow 类型和数据的方式，您应该创建一个 ConvertOptions 实例，并将其传递给 read_csv()

>>> table = csv.read_csv('tips.csv.gz', convert_options=csv.ConvertOptions(
...     column_types={
...         'total_bill': pa.decimal128(precision=10, scale=2),
...         'tip': pa.decimal128(precision=10, scale=2),
...     }
... ))
>>> table
pyarrow.Table
col1: int64
col2: string
----
col1: [[1,2,3]]
col2: [["a","b","c"]]

注意

要将列指定为 duration，CSV 值必须是符合预期单位的数值字符串（例如，当使用 duration[ms] 时，60000 代表 60 秒）。

可用的转换选项包括：

check_utf8	是否检查字符串列的 UTF8 有效性。
column_types	显式地将列名映射到列类型。
null_values	一组表示数据中 null 值的字符串。
true_values	一组表示数据中布尔值 true 的字符串。
false_values	一组表示数据中布尔值 false 的字符串。
decimal_point	浮点数和十进制数据中用作小数点的字符。
timestamp_parsers	与 strptime() 兼容的格式字符串序列，在尝试推断或转换时间戳值时按顺序尝试（也可以使用特殊值 ISO8601()）。
strings_can_be_null	字符串/二进制列是否可以包含空值。
quoted_strings_can_be_null	带引号的值是否可以为空。
auto_dict_encode	是否尝试自动对字符串/二进制数据进行字典编码。
auto_dict_max_cardinality	auto_dict_encode 的最大字典基数。
include_columns	要包含在 Table 中的列名。
include_missing_columns	如果为 false，则当 include_columns 中的列不在 CSV 文件中时会报错。

另请参阅

有关更多示例，请参阅 ConvertOptions。

增量读取

对于内存受限的环境，也可以使用 open_csv() 每次读取一批 CSV 文件。

有几点需要注意：

1. 目前，增量读取器始终是单线程的（无论 ReadOptions.use_threads 如何设置）

2. 类型推断在第一个数据块上进行，之后类型即被锁定；为了确保推断出正确的数据类型，请将 ReadOptions.block_size 设置为足够大的值，或者使用 ConvertOptions.column_types 显式设置所需的数据类型。

字符编码

默认情况下，CSV 文件应采用 UTF8 编码。binary 列接受非 UTF8 数据。可以使用 ReadOptions 类更改编码。

>>> table = csv.read_csv('tips.csv.gz', read_options=csv.ReadOptions(
...    column_names=["n_legs", "entry"],
...    skip_rows=1
... ))
>>> table
pyarrow.Table
n_legs: int64
entry: string
----
n_legs: [[1,2,3]]
entry: [["a","b","c"]]

可用的读取选项包括：

use_threads	是否使用多线程来加速读取。
block_size	从输入流中每次处理多少字节。
skip_rows	在列名（如果有）和 CSV 数据之前要跳过的行数。
skip_rows_after_names	在列名之后要跳过的行数。
column_names	目标表的列名。
autogenerate_column_names	如果 column_names 为空，是否自动生成列名。
encoding	encoding: 对象

另请参阅

有关更多示例，请参阅 ReadOptions。

自定义写入

如果需要写入具有不同约定的 CSV 文件并更改默认设置，您可以创建一个 WriteOptions 实例，并将其传递给 write_csv()

>>> # Omit the header row (include_header=True is the default)
>>> options = csv.WriteOptions(include_header=False)
>>> csv.write_csv(table, "data.csv", options)

增量写入

要每次一批地写入 CSV 文件，请创建一个 CSVWriter。这需要输出（路径或类文件对象）、要写入的数据架构以及上述可选的写入选项。

>>> with csv.CSVWriter("data.csv", table.schema) as writer:
...     writer.write_table(table)

性能

由于 CSV 文件的结构原因，不能指望其性能达到读取 Parquet 等专用二进制格式的水平。尽管如此，Arrow 仍致力于减少读取 CSV 文件的开销。一个合理的预期是，在高性能台式机或笔记本电脑上，每个核心至少能达到 100 MB/s（以源 CSV 字节计，而非目标 Arrow 数据字节）。

性能选项可以通过 ReadOptions 类进行控制。多线程读取是默认设置，旨在实现最高性能，并在所有可用核心上高效分配工作负载。

注意

并发线程数由 Arrow 自动推断。您可以使用 cpu_count() 和 set_cpu_count() 函数分别查看和更改它。