How to make a parquet dataset from an input file through Random access

[jonathan mercier <jo...@cnrgh.fr>]

Dear,

library: pyarrow
version: current stable

I try to read a file supporting random access and convert its data into
a parquet dataset using pyarrow.

Thus I make a pool executor to process input data asynchronously.
Each process read through a stream and return at the end a
pyarrow.lib.Buffer . How to merge all those buffer in order to get one
Table ?
I know how to do it from one buffer but not a collection of buffers:
 -  RecordBatchStreamReader(source).read_all()


My current code :


sink = BufferOutputStream()
writer = RecordBatchStreamWriter(sink, a_schema)

# In order to call write at around 1Mo of data and reuse the buffer
buffer = (
        list(None for _ in range(0, 70)),
        list(None for _ in range(0, 70)),
    )
buffer_index=0

while iterator:
    try:
        buffer[0][buffer_index] = iterator.a        buffer[1][buffer_index] = iterator.b
   except IndexError:
       batch = record_batch([ buffer[0][:buffer_index],
                              buffer[1][:buffer_index]],
                            a_schema)
       writer.write(batch)
       buffer_index=0

if index != 0:
   batch = record_batch([ buffer[0][:buffer_index],
                          buffer[1][:buffer_index]],
                        a_schema)
   writer.write(batch)
writer.close()
return sink.getvalue()



Thanks for your help

best regards

-- 
                Researcher computational biology
                PhD, Jonathan MERCIER
                
                Bioinformatics (LBI)
                2, rue Gaston
                Crémieux
                91057 Evry Cedex
            
            
                Tel :(33) 1 60 87 83 44
                Email :jonathan.mercier@cnrgh.fr
                
            
        
    

Re: How to make a parquet dataset from an input file through Random access

[Wes McKinney <we...@gmail.com>]

If you have used *StreamWriter to create a buffer that you're sending
to another process, if you want to get it back into a record batch or
table, you need to read it with
pyarrow.ipc.open_stream(...).read_all() and then you can concatenate
the resulting tables with pyarrow.concat_tables (or use
Table.from_batches if you have a sequence of record batches)

Hope this helps

On Thu, Jan 21, 2021 at 6:19 AM Jonathan MERCIER
<jo...@cnrgh.fr> wrote:
>
> Same question but more simple to understand.
>
> Using pyarrow and working with pieces of data by process (multi-process
> as workaround GIL limitation). What is the correct way to handle this task ?
>
> 1. each // process have to create create a list of records store them
> into a record batch and return this batch
>
> 2. each // process have to create an output and writer buffer , create a
> list of records store them into a record batch and write this record
> batch into the stream writer. The process return the corresponding buffer ?
>
> with the answer (1) I see how to merge all of those batch but with
> solution (2) how to merge all buffer to one once each process has
> returned their buffer ?
>
>
>
> Thanks
>
>
> --
> Jonathan MERCIER
>
> Researcher computational biology
>
> PhD, Jonathan MERCIER
>
> Centre National de Recherche en Génomique Humaine (CNRGH)
>
> Bioinformatics (LBI)
>
> 2, rue Gaston Crémieux
>
> 91057 Evry Cedex
>
> Tel :(33) 1 60 87 34 88
>
> Email :jonathan.mercier@cnrgh.fr <ma...@cnrgh.fr>
>

Re: How to make a parquet dataset from an input file through Random access

[Jonathan MERCIER <jo...@cnrgh.fr>]

Same question but more simple to understand.

Using pyarrow and working with pieces of data by process (multi-process 
as workaround GIL limitation). What is the correct way to handle this task ?

1. each // process have to create create a list of records store them 
into a record batch and return this batch

2. each // process have to create an output and writer buffer , create a 
list of records store them into a record batch and write this record 
batch into the stream writer. The process return the corresponding buffer ?

with the answer (1) I see how to merge all of those batch but with 
solution (2) how to merge all buffer to one once each process has 
returned their buffer ?



Thanks


-- 
Jonathan MERCIER

Researcher computational biology

PhD, Jonathan MERCIER

Centre National de Recherche en Génomique Humaine (CNRGH)

Bioinformatics (LBI)

2, rue Gaston Crémieux

91057 Evry Cedex

Tel :(33) 1 60 87 34 88

Email :jonathan.mercier@cnrgh.fr <ma...@cnrgh.fr>