Efficient way to move data from one parquet file to another

[avakaluk]

Hi,
I am developing application that optimizes parquet files. As part of the flow I need an efficient way to move data from source to target parquet file. This is especially relevant for large string objects. I use low level API and would like to see if there is suggested way to handle this scenario. Ideally I would prefer that data is copied to target file without being unmarshalled to managed code and marshalled back. IS there any way to achieve this?

Thank you in advance,
Alex

[adamreeve]

Hi Alex

When you read string typed data with the low level API, values are read as an array of ByteArray values. These just store a pointer and length, and point to memory allocated by the C++ Arrow library to read string data into. This is the same type used to write string values with the low-level API, and you can pass these directly back to a column writer without having to marshal the data to .NET strings. You just need to be aware that the ByteArray values are only guaranteed to remain valid until the next call to ReadBatch (or the column reader is disposed), after that the pointers may be invalid or point to data that has been overwritten.

Here's an example of re-writing a Parquet file with string typed data using the low level API:

// Write test file using high level API
{
    var columns = new Column[]
    {
        new Column<string>("s"),
    };
    using var writer = new ParquetFileWriter("test.parquet", columns);
    using var rowGroup = writer.AppendRowGroup();
    using var col = rowGroup.NextColumn().LogicalWriter<string>();
    col.WriteBatch(new [] {"a", "b", "c"});
}

// Read and re-write row groups using low level API, without converting
// byte array values to .NET strings.
{
    using var reader = new ParquetFileReader("test.parquet");
    using var schema = (ParquetSharp.Schema.GroupNode)reader.FileMetaData.Schema.SchemaRoot;
    using var writerPropertiesBuilder = new WriterPropertiesBuilder();
    using var writerProperties = writerPropertiesBuilder.Build();
    using var writer = new ParquetFileWriter("modified.parquet", schema, writerProperties);
    for (var rg = 0; rg < reader.FileMetaData.NumRowGroups; rg++)
    {
        using var readerRowGroup = reader.RowGroup(rg);
        using var writerRowGroup = writer.AppendRowGroup();
        using var readerCol = (ColumnReader<ByteArray>)readerRowGroup.Column(0);
        using var writerCol = (ColumnWriter<ByteArray>)writerRowGroup.NextColumn();

        const int bufferSize = 1024;
        var valuesBuffer = new ByteArray[bufferSize];
        var defLevels = new short[bufferSize];
        var repLevels = new short[bufferSize];

        long valuesRead = 0;
        while (valuesRead < readerCol.ColumnChunkMetaData.NumValues)
        {
            var numLevels = readerCol.ReadBatch(
                valuesBuffer.Length, defLevels, repLevels, valuesBuffer, out var numValues);
            valuesRead += numValues;
            writerCol.WriteBatch(
                (int)numValues,
                defLevels.AsSpan(0, (int)numLevels),
                repLevels.AsSpan(0, (int)numLevels),
                valuesBuffer.AsSpan(0, (int)numValues));
        }
        writerRowGroup.Close();
    }

    writer.Close();
}

// Test reading back with high level API
{
    using var reader = new ParquetFileReader("modified.parquet");
    for (var rg = 0; rg < reader.FileMetaData.NumRowGroups; rg++)
    {
        using var rowGroup = reader.RowGroup(rg);
        using var col = rowGroup.Column(0).LogicalReader<string>();
        var values = col.ReadAll((int) rowGroup.MetaData.NumRows);
        foreach (var value in values)
        {
            Console.WriteLine(value);
        }
    }
}

[avakaluk]

Thank you for your prompt reply.

Your code relies on ColumnChunkMetaData, which is declared as internal, so I used NumRows from RowGrouReader
internal readonly ColumnChunkMetaData ColumnChunkMetaData;

Another small tweak in code is that for columns defines as Optional and that have null values, numValues will return number of not null values and not number of fetched values.

Solution you proposed works in case that Repetition level of source and target column is the same. In my case we can have scenario that source column is with Repetition.Required and target is with Repetition.Optional. What I see is that once I read data from column that is defined as Required defLevels are returned as "0", so if I send fetched data to writer it generates nulls for all values. Is it expected behavior and I need to populate defLevels to "1" on my side or it could be handled on library side?

Alex

[adamreeve]

Ah, possibly ColumnChunkMetaData.NumValues should be exposed publicly somehow to make this possible then, as this has a different meaning to NumRows. ColumnChunkMetaData.NumValues is the number of non-null leaf values (eg. for list typed columns it will be the total number of non-null elements in all lists).

What I see is that once I read data from column that is defined as Required defLevels are returned as "0", so if I send fetched data to writer it generates nulls for all values. Is it expected behavior and I need to populate defLevels to "1" on my side or it could be handled on library side?

Yes this is expected behaviour and is something you would need to handle if you are doing this transformation. For a required column the maximum definition level is 0, but for an optional column it is 1 and 0 indicates null values.

[avakaluk]

Understood. I use combination of numLevels and NumRows to read all data and will manage to Nullable transformation on my side. Thank you for your prompt support.

[avakaluk]

I tried this approach and it works but in my scenario I need to read all column data from row group in order to shuffle it to multiple output files. And it seems that ReadBatch internally enforce batch size that could be smaller than row group size. This means that I need multiple calls to read all data, and according to your post above it is not guaranteed that already returned ByteArrays will remain after I read more data. Is It possible to have an option to read all values, without enforcing batching?

[avakaluk]

@adamreeve - I am using your suggested solution to read data through low level API. It works, but for some reason returns very small number of values/defLevels in every call to ReadBatch. I am requesting 2048 elements, but in most of the cases data is retuned in under 100 items batches. My source files have over million records, so it makes process of moving data to output parquet file very slow. It may rely to fact that I am converting number of files in parallel. Are there any parameters that can help me to tune number of elements that are returned in each batch? Machine I am running the process has enough available memory, so this is not a memory issue.

[adamreeve]

I believe this is limited by the number of values stored in the data pages of the files you're reading. Eg. see this comment here from the Arrow source: https://github.com/apache/arrow/blob/9801801df3f75339f705264252c0eac189f5f2a3/cpp/src/parquet/column_reader.cc#L1116-L1118

There aren't any parameters that would allow you to read from multiple pages at once as far as I know.

If you control how these input files are being written you could try increasing the page size or write batch size. Otherwise you might need to do something like copy data you've read into a temporary buffer before it is written. For string values you can do this by copying the raw bytes without needing to convert them to strings.