在python中将字节块流式传输到csv行

Question

我需要在不完全下载的情况下逐行处理大型远程 CSV。

下面是我得到的最接近的。 我从 Azure 迭代字节块，并有一些代码来处理截断的行。 但是，如果 csv 值包含换行符，这将不起作用，因为我无法区分值换行符和 csv 换行符。

# this does not work
def azure_iter_lines(logger_scope, client, file_path):
    # get a StorageStreamDownloader
    # https://docs.microsoft.com/en-us/python/api/azure-storage-file-datalake/azure.storage.filedatalake.storagestreamdownloader?view=azure-python
    file_client = client.get_file_client(file_path)
    file_handle = file_client.download_file()

    truncated_line = ''
    for chunk in file_handle.chunks():
        # have the previous truncated line appended to the next block
        chunk_txt = truncated_line + chunk.decode("utf-8")
        lines = chunk_txt.split('\n') # THIS CANNOT WORK AS VALUES CONTAIN NEWLINES
        for line in lines[0:len(lines)-2]:
            yield line
        truncated_line = lines[len(lines)-1]

    # process the last chunk (same code)
    chunk_txt = truncated_line
    lines = chunk_txt.split('\n') # THIS CANNOT WORK AS VALUES CONTAIN NEWLINES
    for line in lines[0:len(lines)-2]:
        yield line
    truncated_line = lines[len(lines)-1]

理想情况下我会使用 csv.DictReader() 但我无法这样做，因为它会完全下载文件。

# this does not work
def azure_iter_lines(logger_scope, client, file_path):
    file_client = client.get_file_client(file_path)
    file_handle = file_client.download_file()
    buffer = io.BytesIO()
    file_handle.readinto(buffer) # THIS DOWNLOADS THE FILE ENTIRELY
    csvreader = csv.DictReader(buffer, delimiter=";")
    return csvreader

这是使用@H.Leger 的一些提示的更新

请注意，这仍然不起作用

file_client = client.get_file_client(file_path)
file_handle = file_client.download_file()
stream = codecs.iterdecode(file_handle.chunks(), 'utf-8')
csvreader = csv.DictReader(stream, delimiter=";")
for row in csvreader:
    print(row)
# => _csv.Error: new-line character seen in unquoted field - do you need to open the file in universal-newline mode?

编辑：基于@paiv 答案的最终解决方案

编辑：更新了使用 io 代替编解码器的解决方案以加快解析速度

import io
import csv
import ctypes as ct

# bytes chunk iterator to python stream adapter 
# https://stackoverflow.com/a/67547597/2523414

class ChunksAdapter:
    def __init__(self, chunks):
        self.chunks = chunks
        self.buf = b''
        self.closed = False
    
    def readable(self):
        return True
        
    def writable(self):
        return False
    
    def seekable(self):
        return False
        
    def close(self):
        self.closed = True
        
    def read(self, size):
        if not self.buf:
            self.buf = next(self.chunks, b'')
        res, self.buf = self.buf[:size], self.buf[size:]
        return res



# get the downloader object
file_client = client.get_file_client(file_path)
downloader = file_client.download_file()
# adapt the downloader iterator to a byte stream
file_object = ChunksAdapter(downloader.chunks())
# decode bytes stream to utf-8
text_stream = io.TextIOWrapper(file_object, encoding='utf-8', newline='') 

# update csv field limit to handle large fields
# https://stackoverflow.com/a/54517228/2523414
csv.field_size_limit(int(ct.c_ulong(-1).value // 2)) 

csvreader = csv.DictReader(text_stream, delimiter=";", quotechar='"', quoting=csv.QUOTE_MINIMAL)
for row in csvreader:
    print(row)

Answer 1

免责声明：我对 Azure 的细节知之甚少。最终，您也希望流式传输单独的块。

在 Python 中，给定 file object，您可以这样设置 CSV 流：

import codecs
import csv

codec = codecs.getreader('utf-8')
text_stream = codec(file_object)
csvreader = csv.DictReader(text_stream)

现在您可以迭代 csvreader，它将以流式传输方式从 file_object 读取。

编辑：正如@Martijn Pieters 所建议的，我们可以使用TextIOWrapper 而不是codecs 来获得性能：

text_stream = io.TextIOWrapper(file_object, encoding='utf-8', newline='')

检查the comment in csv module 上的newline 参数。

但是Azure的StorageStreamDownloader没有提供python的文件对象接口。它有 .chunks() 生成器（我假设它会调用单独的 HTTP 请求来检索下一个块）。

您可以使用简单的适配器将.chunks() 适配为文件对象：

class ChunksAdapter:
    def __init__(self, chunks):
        self.chunks = chunks
        self.buf = b''
        
    def read(self, size):
        if not self.buf:
            self.buf = next(self.chunks, b'')
        res, self.buf = self.buf[:size], self.buf[size:]
        return res

并使用喜欢

downloader = file_client.download_file()
file_object = ChunksAdapter(downloader.chunks())

一定要为the appropriate CSV dialect配置DictReader。

并在the blob client 上为max_single_get_size、max_chunk_get_size 设置适当的值。

Answer 2

我相信requests 包对您有用。在获取文件时使用stream 选项和Response.iter_lines() 函数should do what you need：

import codecs
import csv
import requests

url = "https://navitia.opendatasoft.com//explore/dataset/all-datasets/download?format=csv"
r = requests.get(url, stream=True)  # using the stream option to avoid loading everything

try:
    buffer = r.iter_lines()  # iter_lines() will feed you the distant file line by line
    reader = csv.DictReader(codecs.iterdecode(buffer, 'utf-8'), delimiter=';')
    for row in reader:
        print(row)  # Do stuff here
finally:
    r.close()