Export to Parquet and DuckDB

import pyarrow.parquet as pq
from bloklake_arrow import query

def export(body: dict, start: int, end: int, path: str, row_type: str = "logs") -> int:
    writer, frm, rows = None, start, 0
    while frm <= end:
        tables, meta = query({**body, "fromBlock": frm, "toBlock": end})
        table = tables.get(row_type)
        if table is not None and table.num_rows:
            writer = writer or pq.ParquetWriter(path, table.schema, compression="zstd")
            writer.write_table(table)
            rows += table.num_rows
        frm = int(meta["bl.next_block"])
    if writer:
        writer.close()
    return rows

for lo in range(START, END + 1, 100_000):
    hi = min(lo + 99_999, END)
    export(body, lo, hi, f"transfers/blocks={lo:010d}-{hi:010d}.parquet")

-- topics are 32-byte BLOBs: the address is the last 40 hex characters
select '0x' || lower(substr(hex(topic2), 25)) as receiver,
       count(*) as transfers
from 'transfers/*.parquet'
group by 1
order by transfers desc
limit 10;

import pyarrow as pa

def to_decimal(col: pa.ChunkedArray) -> pa.Array:
    return pa.array([None if b is None else int.from_bytes(b, "big") for b in col.to_pylist()],
                    pa.decimal128(38, 0))