Support for streaming/batch processing?

[mcrumiller]

In pandas, batch-processing of a resultset is possible via chunking:

engine = create_engine("postgresql://user:pass@server/db")
conn = engine.connect().execution_options(stream_results=True)

for df_chunked in pd.read_sql(query, conn, chunksize=1000):
    # ...operate on chunk

Is there something similar in connector-x which would enable us to work on a large resultset in batches?

[wangxiaoying]

Hi @mcrumiller , thank you for the suggestion! Indeed, it is desirable to support chunksize and we will definitely consider it in the future plan. But I want to mention here that connectorx is much more efficient than pandas in terms of memory usage. So it can support a much larger result in one batch.

[zafarmah]

Continuing on the with the same discussion, also I would like to ask what are the alternates for utilising this parameter ?

[WilliamStam]

there are soo many ways to handle a dataframe via streaming once "local" but fetching from a db is still a "load everything into memory" type affair which uses a lot of ram the larger the datasets. anyone found solutions for it?

---

for anyone that is curious... i sort of came up with a lame "solution" for sqlalchemy connecting to oracle. this isn't the right place for it since its not using connectorx (i came here looking for streaming options on read_sql).

the idea is to stream in and then for each batch write out a temp file and at the end combine them into the outputed file

def _get_dtype(column):
    if isinstance(column.type, types.String):
        return polars.String
    elif isinstance(column.type, types.Float):
        return float
    elif isinstance(column.type, types.Integer):
        return polars.Int64
    elif isinstance(column.type, types.TIMESTAMP):
        return polars.Datetime
    elif isinstance(column.type, types.DateTime):
        return polars.Datetime
    elif isinstance(column.type, types.Date):
        return polars.Date
    elif isinstance(column.type, types.Boolean):
        return bool
    elif isinstance(column.type, Decimal):
        return polars.Decimal
    elif isinstance(column.type, types.Numeric):
        return polars.Decimal

    # Catch all type - handle provider specific types in another elif block
    logger.debug(f"unknown type for {column.name}: {column.type} ({type(column.type)})")
    return polars.String

files = []

async def _write_temp(records=None, dataframe_schema=None, filename: str = None):
    if dataframe_schema is None:
        dataframe_schema = {}

    if records is None:
        records = []

    bdf = polars.LazyFrame(records, schema=dataframe_schema, orient="row")

    filename = f"{filename}"

    bdf.sink_parquet(filename,compression='snappy')
    files.append(filename)

    del bdf
    del records


selectable = select(MyBigTable)
schema = dict()
for column in selectable.selected_columns:
    schema[str(column.name)] = _get_dtype(column)

result = await (session.stream(selectable.execution_options(yield_per=1000)))

temp_data = []
temp_files = "tmp"

batch_time = time.time()
batch_number = 0
async for row in result.scalars():
  temp_data.append(tuple(getattr(row, x) for x in schema.keys()))

  if len(temp_data) >= batch_size:
      f = await _write_temp(temp_data, dataframe_schema=schema, filename=f"{temp_files}-{batch_number}")
      temp_data = []
      batch_time = time.time()
      batch_number = batch_number + 1

await _write_temp(temp_data, dataframe_schema=schema, filename=f"{temp_files}-{batch_number}")


polars.scan_parquet(files).sink_parquet(f"combined.parquet")