pyarrow read_csv - how to fill trailing optional columns with nulls

user_16499523 0
Viewed 505

I can't find an option or workaround for this using pyarrow.csv.read_csv and there are many other reasons why using pandas doesn't work for us.

We have csv files with final columns that are effectively optional, and the source data doesn't always include empty cells for them, for example:

name,date,serial_number,prior_name,comments
A,2021-01-01,1234
B,2021-01-02,1235,A,Name changed for new version
C,2021-01-02,1236,B

This fails with an error like pyarrow.lib.ArrowInvalid: CSV parse error: Expected 5 columns, got 3:

I've got to assume that pyarrow can handle this, but I can't see how. Even the invalid row handler doesn't appear to let me return the "appropriate" value, only to "skip" these rows. That would even be okay if I could save them and append later, but as arrow tables are immutable, it just seems like there should be a more straightforward way to handle these cases.

apache-arrow pyarrow csv parsing
1 Answers

As Pace noted, this is unfortunately not presently available in pyarrow. We actually process the "csv" files on the fly after extracting from a zip, and so creating intermediate files wasn't an option.

For anyone else needing a quick-ish way to handle this, I was able to get around this (and a couple other issues) by creating a class to wrap the stream and overwrite read(size, *args, **kwargs) to quickly perform the stripping. Even with the middleman class, it's faster than attempting to load in pandas (and there are several other reasons why we aren't using pandas here).

Here's a template example:

class StreamWrapper():
    
    def __init__(self, obj=None):
        self.delegation = obj
        self.header = None

    def __getattr__(self, *args, **kwargs):
        # any other call is delegated to the stream/object
        return(self.delegation.__getattribute__(*args, **kwargs))

    def read(self, size=None, *args, **kwargs):
        bytedata = self.delegation.read(size, *args, **kwargs)          
        # .. the rest of the logic pre-processes the byte data, 
        # identifies the header and number of columns (which are retained persistently),
        # and then strips out extra columns when encountered
        return(bytedata)

This allow our call to be:

    df = pyarrow.csv.read_csv(
        StreamWrapper(zipfile_stream),
        parse_options = ...
    )
user_164995230
Related