如何使用 BytesIO 作为 pandas.read_csv 数据源

Question

我正在尝试使用 pandas.read_csv(bytes, chunksize=n) 执行 csv 数据解析，其中 bytes 是我想从数据库 CLOB 字段接收的持续数据流，按块读取。

reader = pandas.read_csv(io.BytesIO(b'1;qwer\n2;asdf\n3;zxcv'), sep=';', chunksize=2)
for row_chunk in reader:
  print(row_chunk)

上面的代码工作正常，但我想使用一些可更新的流而不是固定的io.BytesIO(b'...') 我试图重新定义这样的读取方法

class BlobIO(io.BytesIO):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._chunk_size = 4
        self._file_data_table = 'my_table'
        self._job_id = 'job_id'
        self._get_raw_sql = """
            select dbms_lob.substr(body, {0}, {1})
            from {2}
            where job_id = '{3}'
        """
        dsn_tns = cx_Oracle.makedsn('host', 'port', 'service_name')
        self.ora_con = cx_Oracle.connect('ora_user', 'ora_pass', dsn_tns)
        self.res = b''
        self.ora_cur = self.ora_con.cursor()
        self.chunker = self.get_chunk()
        next(self.chunker)

    def get_chunk(self):
        returned = 0
        sended = (yield)
        self._chunk_size = sended or self._chunk_size
        while True:
           to_exec = self._get_raw_sql.format(
               self._chunk_size,
               returned + 1,
               self._file_data_table,
               self._job_id)
           self.ora_cur.execute(to_exec)
           self.res = self.ora_cur.fetchall()[0][0]
           returned += self._chunk_size
           yield self.res
           sended = (yield self.res)
           self._chunk_size = sended or self._chunk_size
           if not self.res:
               break

    def read(self, nbytes=None):
        if nbytes:
            self.chunker.send(nbytes)
        else:
            self.chunker.send(self._chunk_size)
        try:
            to_return = next(self.chunker)
        except StopIteration:
            self.ora_con.close()
            to_return = b''
        return to_return

buffer = BlobIO()
reader = pandas.read_csv(buffer, encoding='cp1251', sep=';', chunksize=2)

但看起来我做错了什么，因为pd.read_csv 从未在最后一行被执行，我不明白那里发生了什么。

也许创建buffer = BytesIO(b'')，然后将新数据写入缓冲区buffer.write(new_chunk_from_db) 可能是一种更好的方法，但我不明白何时应该调用这样的写入操作。

我相信我可以创建一个包含 CLOB 内容的临时文件，然后我可以将其传递给 read_csv，但我真的很想跳过这一步，直接从数据库中读取数据。

请给我一些指示。

Answer 1

cx_Oracle 提供读取 LOB 的本地方式。似乎用 cx_Oracle LOB 读取覆盖 BytesIO 读取就可以了：

class BlobIO(BytesIO):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.res = b''
        self.ora_con = db.get_conn()
        self.ora_cur = self.ora_con.cursor()
        self.ora_cur.execute("select lob from table")
        self.res = self.ora_cur.fetchall()[0][0]
        self.offset = 1

    def read(self, size=None):
        r = self.res.read(self.offset, size)
        self.offset += size
        # size + 1 should be here to perform nonoverlaping reads
        # but looks like panadas C parser uses some kind of overlaping
        # because while testing size+1 - parser occasionally missed some bytes
        if not r:
            self.ora_cur.close()
            self.ora_con.close()
        return r

blob_buffer = BlobIO()
reader = pandas.read_csv(
        blob_buffer, 
        chunksize=JobContext.rchunk_size)
for row_chunk in reader:
    print(row_chunk)