Передача SFTPFile-соединения Paramiko в качестве входных данных для dask.dataframe.read_parquet

Question

Я пытался передать класс paramiko.sftp_file.SFTPFile вместо URL файла для pandas.read_parquet, и он работал нормально.Но когда я попробовал то же самое с Dask, он выдал ошибку.Ниже приведен код, который я пытался запустить, и ошибка, которую я получил.Как я могу сделать эту работу?

import dask.dataframe as dd
import parmiko
ssh=paramiko.SSHClient()
sftp_client = ssh.open_sftp()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
source_file=sftp_client.open(str(parquet_file),'rb')
full_df = dd.read_parquet(source_file,engine='pyarrow')
print(len(full_df))

Traceback (most recent call last):
  File "C:\Users\rrrrr\Documents\jackets_dask.py", line 22, in <module>
    full_df = dd.read_parquet(source_file,engine='pyarrow')
  File "C:\Users\rrrrr\AppData\Local\Programs\Python\Python37\lib\site-packages\dask\dataframe\io\parquet.py", line 1173, in read_parquet
    storage_options=storage_options
  File "C:\Users\rrrrr\AppData\Local\Programs\Python\Python37\lib\site-packages\dask\bytes\core.py", line 368, in get_fs_token_paths
    raise TypeError('url type not understood: %s' % urlpath)
TypeError: url type not understood: <paramiko.sftp_file.SFTPFile object at 0x0000007712D9A208>

Answer 1

Ситуация изменилась, и вы можете сделать это прямо сейчас с помощью Dask.Пастерный ответ от Можно ли прочитать .csv с удаленного сервера, используя совместно метод Paramiko и Dask read_csv ()?

В основной версии Dask операции с файловой системойтеперь используют fsspec, который наряду с предыдущими реализациями (s3, gcs, hdfs) теперь поддерживает некоторые дополнительные файловые системы , см. сопоставление с идентификаторами протоколов fsspec.registry.known_implementations.

Короче говоря, использование URL-адреса типа "sftp: // user: pw @ host: port / path" теперь должно работать для вас, если вы устанавливаете fsspec и Dask из master.

Answer 2

Dask напрямую не поддерживает файловые объекты.

Вам потребуется реализовать их интерфейс "файловой системы" .

Я не уверен, какой минимальный набор методов нужно реализовать, чтобы разрешить read_parquet. Но вам определенно нужно реализовать open. Примерно так:

class SftpFileSystem(object):
    def open(self, path, mode='rb', **kwargs):
        return sftp_client.open(path, mode)

dask.bytes.core._filesystems['sftp'] = SftpFileSystem

df = dd.read_parquet('sftp://remote/path/file', engine='pyarrow')

Полный набор методов см. В реализации LocalFileSystem.

На самом деле есть реализация такой файловой системы для SFTP в библиотеке fsspec:
https://filesystem -spec.readthedocs.io / ен / последний / api.html # fsspec.implementations.sftp.SFTPFileSystem

См. Также Можно ли читать .csv с удаленного сервера, используя совместно метод read_csv () от Paramiko и Dask?