使用pyarrow读取FTP分区Parquet文件失败，fastparquet正常

文件结构 #

FTP服务器上存储的分区Parquet文件结构如下：

├───train_set=TS04
│       part.0.parquet
│       part.1.parquet
│       part.2.parquet
│
├───train_set=TS05
│       part.10.parquet
│       part.11.parquet
│
├───train_set=TS06
│       part.12.parquet
│       part.13.parquet
│
├───train_set=TS07
│       part.17.parquet
│
├───train_set=TS08
│       part.20.parquet
│
└───train_set=TS09
        part.21.parquet
        part.22.parquet

fastparquet引擎正常读取 #

使用fastparquet引擎可正常读取整个数据集：

df = pd.read_parquet(
    'ftp://user@192.168.1.85:21/root_evts',
    storage_options={'password':'123'},
    engine="fastparquet"
)
len(df)  # => 9510029

pyarrow引擎报错 #

切换到pyarrow引擎时，卡顿30秒后抛出错误：

df = pd.read_parquet(
    'ftp://user@192.168.1.85:21/root_evts',
    storage_options={'password':'123'},
    engine="pyarrow"
)

错误信息：

error_reply: 200 Switching to Binary mode.

完整回溯信息：

---------------------------------------------------------------------------
error_reply                               Traceback (most recent call last)
Cell In[15], line 1
----> 1 o_jru = pd.read_parquet(
      2     'ftp://user@192.168.1.85:21/root_evts',
      3     storage_options={'password':'123'},
      4     engine="pyarrow"
      5 )

File ~\source\repos\hotline_tool_venv\Lib\site-packages\pandas\io\parquet.py:670, in read_parquet(path, engine, columns, storage_options, use_nullable_dtypes, dtype_backend, filesystem, filters, **kwargs)
    667     use_nullable_dtypes = False
    668 check_dtype_backend(dtype_backend)
---> 670 return impl.read(
    671     path,
    672     columns=columns,
    673     filters=filters,
    674     storage_options=storage_options,
    675     use_nullable_dtypes=use_nullable_dtypes,
    676     dtype_backend=dtype_backend,
    677     filesystem=filesystem,
    678     **kwargs,
    679 )

File ~\source\repos\hotline_tool_venv\Lib\site-packages\pandas\io\parquet.py:272, in PyArrowImpl.read(self, path, columns, filters, use_nullable_dtypes, dtype_backend, storage_options, filesystem, **kwargs)
    265 path_or_handle, handles, filesystem = _get_path_or_handle(
    266     path,
    267     filesystem,
    268     storage_options=storage_options,
    269     mode="rb",
    270 )
    271 try:
---> 272     pa_table = self.api.parquet.read_table(
    273         path_or_handle,
    274         columns=columns,
    275         filesystem=filesystem,
    276         filters=filters,
    277         **kwargs,
    278     )
    279     result = pa_table.to_pandas(**to_pandas_kwargs)
    281     if manager == "array":

File ~\source\repos\hotline_tool_venv\Lib\site-packages\pyarrow\parquet\core.py:1825, in read_table(source, columns, use_threads, schema, use_pandas_metadata, read_dictionary, memory_map, buffer_size, partitioning, filesystem, filters, use_legacy_dataset, ignore_prefixes, pre_buffer, coerce_int96_timestamp_unit, decryption_properties, thrift_string_size_limit, thrift_container_size_limit, page_checksum_verification)
   1813     # TODO test that source is not a directory or a list
   1814     dataset = ParquetFile(
   1815         source, read_dictionary=read_dictionary,
   1816         memory_map=memory_map, buffer_size=buffer_size,
   (...)
   1822         page_checksum_verification=page_checksum_verification,
   1823     )
-> 1825 return dataset.read(columns=columns, use_threads=use_threads,
   1826                     use_pandas_metadata=use_pandas_metadata)

File ~\source\repos\hotline_tool_venv\Lib\site-packages\pyarrow\parquet\core.py:1468, in ParquetDataset.read(self, columns, use_threads, use_pandas_metadata)
   1460         index_columns = [
   1461             col for col in _get_pandas_index_columns(metadata)
   1462             if not isinstance(col, dict)
   1463         ]
   1464         columns = (
   1465             list(columns) + list(set(index_columns) - set(columns))
   1466         )
-> 1468 table = self._dataset.to_table(
   1469     columns=columns, filter=self._filter_expression,
   1470     use_threads=use_threads
   1471 )
   1473 # if use_pandas_metadata, restore the pandas metadata (which gets
   1474 # lost if doing a specific `columns` selection in to_table)
   1475 if use_pandas_metadata:

File ~\source\repos\hotline_tool_venv\Lib\site-packages\pyarrow\_dataset.pyx:562, in pyarrow._dataset.Dataset.to_table()

File ~\source\repos\hotline_tool_venv\Lib\site-packages\pyarrow\_dataset.pyx:3722, in pyarrow._dataset.Scanner.to_table()

File ~\source\repos\hotline_tool_venv\Lib\site-packages\pyarrow\error.pxi:154, in pyarrow.lib.pyarrow_internal_check_status()

File ~\source\repos\hotline_tool_venv\Lib\site-packages\pyarrow\types.pxi:88, in pyarrow.lib._datatype_to_pep3118()

File ~\source\repos\hotline_tool_venv\Lib\site-packages\fsspec\spec.py:1846, in AbstractBufferedFile.read(self, length)
   1843 if length == 0:
   1844     # don't even bother calling fetch
   1845     return b""
-> 1846 out = self.cache._fetch(self.loc, self.loc + length)
   1847 self.loc += len(out)
   1848 return out

File ~\source\repos\hotline_tool_venv\Lib\site-packages\fsspec\caching.py:189, in ReadAheadCache._fetch(self, start, end)
    187     part = b""
    188 end = min(self.size, end + self.blocksize)
-> 189 self.cache = self.fetcher(start, end)  # new block replaces old
    190 self.start = start
    191 self.end = self.start + len(self.cache)

File ~\source\repos\hotline_tool_venv\Lib\site-packages\fsspec\implementations\ftp.py:326, in FTPFile._fetch_range(self, start, end)
    323         raise TransferDone
    325 try:
-> 326     self.fs.ftp.retrbinary(
    327         f"RETR {self.path}",
    328         blocksize=self.blocksize,
    329         rest=start,
    330         callback=callback,
    331     )
    332 except TransferDone:
    333     try:
    334         # stop transfer, we got enough bytes for this block

File ~\AppData\Local\Programs\Python\Python311\Lib\ftplib.py:436, in FTP.retrbinary(self, cmd, callback, blocksize, rest)
    422 """Retrieve data in binary mode.  A new port is created for you.
    423 
    424 Args:
   (...)
    433   The response code.
    434 """
    435 self.voidcmd('TYPE I')
-> 436 with self.transfercmd(cmd, rest) as conn:
    437     while 1:
    438         data = conn.recv(blocksize)

File ~\AppData\Local\Programs\Python\Python311\Lib\ftplib.py:393, in FTP.transfercmd(self, cmd, rest)
    391 def transfercmd(self, cmd, rest=None):
    392     """Like ntransfercmd() but returns only the socket."""
-> 393     return self.ntransfercmd(cmd, rest)[0]

File ~\AppData\Local\Programs\Python\Python311\Lib\ftplib.py:353, in FTP.ntransfercmd(self, cmd, rest)
    351 size = None
    352 if self.passiveserver:
-> 353     host, port = self.makepasv()
    354     conn = socket.create_connection((host, port), self.timeout,
    355                                     source_address=self.source_address)
    356     try:

File ~\AppData\Local\Programs\Python\Python311\Lib\ftplib.py:327, in FTP.makepasv(self)
    325 """Internal: Does the PASV or EPSV handshake -> (address, port)"""
    326 if self.af == socket.AF_INET:
-> 327     untrusted_host, port = parse227(self.sendcmd('PASV'))
    328     if self.trust_server_pasv_ipv4_address:
    329         host = untrusted_host

File ~\AppData\Local\Programs\Python\Python311\Lib\ftplib.py:839, in parse227(resp)
    835 '''Parse the '227' response for a PASV request.
    836 Raises error_proto if it does not contain '(h1,h2,h3,h4,p1,p2)'
    837 Return ('host.addr.as.numbers', port#) tuple.'''
    838 if resp[:3] != '227':
-> 839     raise error_reply(resp)
    840 global _227_re
    841 if _227_re is None:

error_reply: 200 Switching to Binary mode.

已排查情况 #

先后测试了Filezilla、garethflowers/docker-ftp-server、delfer/docker-alpine-ftp-server三种FTP服务器实现，均出现相同错误。由于需要使用read_parquet的行过滤功能（仅支持pyarrow引擎），需解决该问题。

排查步骤 #

• 调整FTP连接模式：在storage_options中添加'passive': False，强制使用主动PORT模式，避免PASV模式下的响应解析问题
• 对比FTP命令交互：用抓包工具记录fastparquet和pyarrow读取时的FTP命令序列，重点查看发送PASV命令前后的服务器响应，确认是否有未处理的响应导致解析错误
• 验证fsspec版本：尝试升级fsspec到最新版本，或降级到与fastparquet兼容的版本，排查是否为fsspec的FTP实现bug
• 手动模拟FTP操作：用ftplib编写脚本，复现pyarrow触发的FTP流程（登录→切换二进制模式→发送PASV→读取文件），定位命令序列中的问题
• 测试单个文件读取：直接读取单个part*.parquet文件，而非整个目录，排查是目录遍历逻辑还是文件读取逻辑导致的错误
• 切换pyarrow数据集模式：在read_parquet中添加use_legacy_dataset=True参数，使用旧版数据集读取逻辑，看是否能绕过问题

内容的提问来源于stack exchange，提问作者Arthur Attout