DEV Community: Jerónimo López

The Carpet feature that nobody will use

Jerónimo López — Thu, 15 May 2025 12:02:07 +0000

This week I released a new version of Carpet, the Java library for working with Parquet files. In this version, I've added a feature that I believe nobody will ever use: the ability to read and write BSON-type columns.

Parquet supports embedded types, which allow defining complex data types within a column. Each of these types has its own internal representation as a byte array.

Historically, the Parquet format has defined JSON and BSON types:

JSON, with binary value (JSON) in the schema, represents data as a UTF-8 text (String) serialization of a JSON object. The difference from storing the same content in a STRING type column is that it explicitly defines the content as a JSON object, not just plain text.
BSON, with binary value (BSON) in the schema, represents data as a byte array following the BSON specification. By typing it this way, it also explicitly identifies the content as a BSON object, rather than just any byte array.

In my limited experience with Parquet, I haven't come across examples where these embedded types are used, and no one had requested support for them in Carpet.

So why did I decide to implement support for them? The answer is simple: to challenge myself and see if I could do it, and to test whether the codebase was flexible enough to support these features without breaking anything or imposing limitations on Carpet.

Another reason is that new embedded types have recently been defined in the Parquet specification: VARIANT, GEOMETRY, and GEOGRAPHY. There are still no functional implementations of these new types in Parquet, but they are already defined and work is underway on their implementation in at least two languages. It's wise to lay the groundwork for their future availability.

The Implementation Problem

The key challenge when implementing these types was deciding which Java types to use for representation.

What's the best Java class to represent JSON? Should I create a Carpet-specific class? Use the JSONObject class from the org.json library? Or perhaps the JsonNode class from Jackson? Could I use any object and serialize it to JSON? And if so, with which library? Any of these approaches would tie Carpet to a specific implementation and add unnecessary dependencies for 99.99% of use cases, while also forcing users into a particular library.

The same issue applies to BSON. I could use the MongoDB bson library, but that would create a dependency on a specific implementation that virtually no one needs.

Instead, I opted to use Java's String type to represent JSON content and Parquet's Binary class for BSON. This approach keeps Carpet independent of any specific implementation, allowing users to choose their preferred library for handling the content.

The trade-off is that anyone using this functionality (which I suspect will be very few people) will need to declare their attributes as String or Binary instead of using their own business classes, and will need to handle serialization and deserialization themselves before using Carpet:

record ProductEvent(
    long id,
    String name,
    String jsonData,
    Binary bsonData,
    Instant timestamp) {
}

Annotating Java Types

I couldn't simply use String for JSON and Binary for BSON because Java's String type already automatically maps to Parquet's STRING type. Additionally, Binary needs to support multiple Parquet logical types (STRING, BSON, and eventually VARIANT, GEOMETRY, and GEOGRAPHY).

To solve this, I created annotations for Java record attributes to specify which Parquet data type they should map to.

When you annotate a String attribute with @ParquetJson or a Binary attribute with @ParquetBson, you're telling Carpet that these attributes contain JSON or BSON data respectively, not just regular text or byte arrays.

This record, when written to a Parquet file:

record ProductEvent(
    long id,
    String name,
    @ParquetJson String jsonData,
    @ParquetBson Binary bsonData,
    Instant timestamp) {
}

will generate the following Parquet schema:

message ProductEvent {
    required int64 id;
    optional binary name (STRING);
    optional binary jsonData (JSON);
    optional binary bsonData (BSON);
    required int64 timestamp (TIMESTAMP(MILLIS,true));
}

After introducing annotations for changing attribute types in the Parquet schema, I expanded this capability beyond just embedded types to include Java's String and Enum types as well.

I added the @ParquetString and @ParquetEnum annotations to modify the Parquet logical type of Java attributes of type String, Enum, or Binary. This is particularly useful when conforming to a third-party contract while still using the most convenient data types in your Java code.

This record, when written to a Parquet file:

record ProductEvent(
    long id,
    String name,
    @ParquetString Binary productCode,
    @ParquetString MyEnum category,
    @ParquetEnum String type){
}

will generate the following Parquet schema:

message ProductEvent {
    required int64 id;
    optional binary name (STRING);
    optional binary productCode (STRING);
    optional binary category (STRING);
    optional binary type (ENUM);
}

If you want to know more about the new features, you can read the Carpet documentation on Java type annotations.

Conclusion

While I doubt anyone will use the JSON and BSON functionality in Carpet, developing this feature has been valuable for testing the flexibility of Carpet's codebase and preparing it for the new embedded types being defined in the Parquet specification.

As a bonus, I've enhanced support for Parquet's Binary type and added the ability to modify logical types of attributes in the Parquet schema, capabilities I hadn't originally planned but that could prove useful in certain scenarios.

With this new functionality requiring documentation, I took the opportunity to create a dedicated documentation site for Carpet, moving all the content from the GitHub repository's README to this new resource.

The documentation is built with MkDocs and hosted on GitHub Pages. You can browse the complete documentation at carpet.jerolba.com.

The two versions of Parquet

Jerónimo López — Thu, 20 Feb 2025 06:00:00 +0000

A few days ago, the creators of DuckDB wrote the article: Query Engines: Gatekeepers of the Parquet File Format, which explained how the engines that process Parquet files as SQL tables are blocking the evolution of the format. This is because those engines are not fully supporting the latest specification, and without this support, the rest of the ecosystem has no incentive to adopt it.

In my experience, this issue is not limited to Query Engines but extends to the tools within the ecosystem. Soon after releasing the first version of Carpet, I discovered that there was a version 2 of the format and that the core Java Parquet library does not activate it by default. Since the specification had been finalized for some time, I decided that the best approach was to make Carpet use version 2 by default.

A week later, I discovered at work the hard way that if you are not up to date with Pandas in Python, you cannot read files written with version 2. I had to rollback the change immediately.

Parquet Version 2

Upon researching the topic, you'll find that even though the format specification is finalized, it is not fully implemented across the ecosystem. Ideally, the standard would be whatever the specification defines, but in reality, there is no agreement on the minimum set of features an implementation must support to be considered compatible with version 2.

In this Pull Request from the project that describes the file format, there has been an ongoing discussion for four years about what constitutes the core, and there are no signs of a resolution anytime soon. Reading this other thread on the mailing list, I came to the conclusion that although they are part of the specification, two concepts are mixed that could evolve independently:

Given a series of values in a column, how to encode them efficiently. Being able to incorporate new encodings such as RLE_DICTIONARY or DELTA_BYTE_ARRAY, which further improve compression.
Given an encoded column’s data, where to write it within the file along with its metadata such as headers, nulls, or statistics, which helps to maximize the available metadata while minimizing its size and the number of file reads. This is what they call Data Page V2.

Many would likely prefer to prioritize improvements in encoding over page structure. Finding a file that uses an unknown encoding would make a column unreadable, but a change in how pages are structured would make the entire file unreadable.

What I came to understand is that new logical types are not tied to a specific format version. On the one hand, there are the primitive types that are fixed, but on top of them, logical types are defined: a date is a representation of an int64, a Big Decimal or String is represented with a BYTE_ARRAY. Now the VARIANT type is being defined and I have not seen it associated with either of the two versions.

Meanwhile, in the Machine Learning world, Parquet and ORC have become limiting, requiring specialized features such as handling files with thousands of columns. To address this, two new formats have emerged: Nimble from Facebook and LV2 from LanceDB.

If you want to delve deeper into the topic, I recommend this introductory article. I consider these two formats to be niche solutions and Parquet will continue to reign in the world of data engineering.

Performance of Version 2

The DuckDB article prompted me to investigate the performance implications of Parquet Version 2, which I hadn't considered in [my previous post on compression algorithms]](https://dev.to/jerolba/compression-algorithms-in-parquet-java-4kec).

Configuring file writing with version 2 is straightforward, requiring only a property setting in the writer’s builder:

CarpetWriter<T> writer = new CarpetWriter.Builder<>(outputFile, clazz)
    .withWriterVersion(WriterVersion.PARQUET_2_0)
    .build();

File Size

Italian government dataset:

Format	Version 1	Version 2	Improvement
CSV	1761 MB	1761 MB	-
UNCOMPRESSED	564 MB	355 MB	37 %
SNAPPY	220 MB	198 MB	10 %
GZIP	146 MB	138 MB	5 %
ZSTD	148 MB	144 MB	2 %
LZ4_RAW	209 MB	192 MB	8 %
LZO	215 MB	195 MB	9 %

New York taxi dataset:

Format	Version 1	Version 2	Improvement
CSV	2983 MB	2983 MB	-
UNCOMPRESSED	760 MB	511 MB	33 %
SNAPPY	542 MB	480 MB	11 %
GZIP	448 MB	444 MB	1 %
ZSTD	430 MB	444 MB	-3 %
LZ4_RAW	547 MB	482 MB	12 %
LZO	518 MB	479 MB	7 %

The new encodings in Version 2 compact the data more effectively before compression, resulting in a larger relative improvement for the UNCOMPRESSED version. This leaves less room for subsequent compression algorithms to further reduce the file size (or even slightly worsening it like ZSTD).

Writing

Italian government dataset (seconds):

Format	Version 1	Version 2	Improvement
UNCOMPRESSED	25.0	23.6	6 %
SNAPPY	25.2	23.5	7 %
GZIP	39.3	35.8	9 %
ZSTD	27.3	25.7	6 %
LZ4_RAW	24.9	23.8	4 %
LZO	26.0	24.6	5 %

New York taxi dataset (seconds):

Format	Version 1	Version 2	Improvement
UNCOMPRESSED	57.9	50.2	13 %
SNAPPY	56.4	50.7	10 %
GZIP	91.1	66.9	27 %
ZSTD	64.1	57.1	11 %
LZ4_RAW	56.5	50.5	11 %
LZO	56.1	51.1	9 %

The improvement in writing times is remarkable, but especially in the New York taxi dataset, with a majority of numeric values. Especially noteworthy is the improvement in GZIP format times.

Reading

Italian government dataset (seconds):

Format	Version 1	Version 2	Improvement
UNCOMPRESSED	11.4	11.3	1 %
SNAPPY	12.5	11.5	8 %
GZIP	13.6	12.8	6 %
ZSTD	13.1	12.2	7 %
LZ4_RAW	12.8	11.3	12 %
LZO	13.1	12.1	7 %

New York taxi dataset (seconds):

Format	Version 1	Version 2	Improvement
UNCOMPRESSED	37.4	33.0	12 %
SNAPPY	39.9	34.0	15 %
GZIP	40.9	34.4	16 %
ZSTD	41.5	34.1	18 %
LZ4_RAW	41.5	33.6	19 %
LZO	41.1	33.7	18 %

In reading, we again see a notable improvement, but even better in the taxi dataset with many decimal types.

Conclusion

Although this post might seem like a critique of Parquet, that is not my intention. I am simply documenting what I have learned and explaining the challenges maintainers of an open format face when evolving it. All the benefits and utilities that a format like Parquet has far outweigh these inconveniences.

The improvements that the latest version of Parquet brings help reduce file sizes and processing times, but the difference is not dramatic. Given the low adoption of Version 2 in the ecosystem, for now these improvements do not help to justify potential compatibility problems when you integrate with third parties. However, if you control all parts of the process, consider adopting the latest specification.

Most of what I have written is my interpretation, and I could be wrong. If you have better sources or a different opinion, feel free to share it in the comments.

Compression algorithms in Parquet Java

Jerónimo López — Mon, 20 Jan 2025 09:06:16 +0000

Apache Parquet is a columnar storage format optimized for analytical workloads, though it can also be used to store any type of structured data solving multiple use cases.

One of its most notable features is the ability to efficiently compress data using different compression techniques at two stages of its process. This reduces storage costs and improves reading performance.

This article explains file compression in Parquet for Java, provides usage examples, and analyzes its performance.

Compression Techniques

Unlike traditional row-based storage formats, Parquet uses a columnar approach, allowing the usage of more specific and effective compression techniques based on data locality and redundancy of values of the same type.

Parquet writes information in binary and applies compression at two distinct levels, using different techniques at each:

While writing the values of a column, it adaptively chooses the encoding type based on the characteristics of the initial values: Dictionary, Run-Length Encoding Bit-Packing, Delta Encoding, etc.
Every time a certain amount of bytes is reached (1MB by default) a page is formed, and the binary block is compressed with the algorithm configured by the programmer (none, GZip, Snappy, LZ4, etc.).

Although the compression algorithm is configured at the file level, the encoding of each column is automatically selected using an internal heuristic (at least in the parquet-java implementation).

The performance of different compression techniques will depend heavily on your data, so there's no silver bullet that guarantees the fastest processing time and lowest space consumption. You will need to execute your own tests.

Code

The configuration is straightforward, and it only needs to be explicitly set when writing. When reading a file, Parquet discovers which compression algorithm was used and applies the corresponding decompression algorithm.

Configuring the Algorithm or Codec

In both Carpet and Parquet with Protocol Buffers and Avro, to configure the compression algorithm, you just need to call the withCompressionCodec method of the builder:

Carpet

CarpetWriter<T> writer = new CarpetWriter.Builder<>(outputFile, clazz)
    .withCompressionCodec(CompressionCodecName.ZSTD)
    .build();

Avro

ParquetWriter<Organization> writer = AvroParquetWriter.<Organization>builder(outputFile)
    .withSchema(new Organization().getSchema())
    .withCompressionCodec(CompressionCodecName.ZSTD)
    .build();

Protocol Buffers

ParquetWriter<Organization> writer = ProtoParquetWriter.<Organization>builder(outputFile)
    .withMessage(Organization.class)
    .withCompressionCodec(CompressionCodecName.ZSTD)
    .build();

The value must be one of the available ones in the CompressionCodecName enum: UNCOMPRESSED, SNAPPY, GZIP, LZO, BROTLI, LZ4, ZSTD, and LZ4_RAW (LZ4 is deprecated, and you should use LZ4_RAW).

Compression Level

Some compression algorithms offer a way to fine-tune the compression level. This level is usually related to the effort they need to apply to find repetition patterns, and the higher the compression, the more time and memory is required for the compression process.

Although they come with a default value, it is modifiable using Parquet's generic configuration mechanism, although each codec uses a different key.

Additionally, the value to choose is not standard and depends on each codec, so you must refer to the documentation of each algorithm to understand what each level offers.

ZSTD

To reference the configuration of the level, ZSTD codec declares a constant: ZstandardCodec.PARQUET_COMPRESS_ZSTD_LEVEL.

Possible values range from 1 to 22, and the default value is 3.

ParquetWriter<Organization> writer = ProtoParquetWriter.<Organization>builder(outputFile)
    .withMessage(Organization.class)
    .withCompressionCodec(CompressionCodecName.ZSTD)
    .config(ZstandardCodec.PARQUET_COMPRESS_ZSTD_LEVEL, "6")
    .build();

LZO

To reference the configuration of the level, LZO codec declares a constant: LzoCodec.LZO_COMPRESSION_LEVEL_KEY.

Possible values range from 1 to 9, 99, and 999, and the default value is "999".

ParquetWriter<Organization> writer = ProtoParquetWriter.<Organization>builder(outputFile)
    .withMessage(Organization.class)
    .withCompressionCodec(CompressionCodecName.LZO)
    .config(LzoCodec.LZO_COMPRESSION_LEVEL_KEY, "99")
    .build();

GZIP

It does not declare any constant, and you have to use the string "zlib.compress.level" directly, with possible values ranging from 0 to 9 and with a default value of "6"

ParquetWriter<Organization> writer = ProtoParquetWriter.<Organization>builder(outputFile)
    .withMessage(Organization.class)
    .withCompressionCodec(CompressionCodecName.GZIP)
    .config("zlib.compress.level", "9")
    .build();

Performance Tests

To analyze the performance of different compression algorithms, I will use two public datasets containing different types of data:

New York City Taxi Trips: with a large number of numeric values and few string values in a few columns. It has 23 columns and contains 19.6 million records.
Cohesion Projects of the Italian Government: many columns with float values and a large quantity and variety of text strings. It has 91 columns and contains 2 million rows.

I will evaluate some of the compression algorithms enabled in Parquet Java: UNCOMPRESSED, SNAPPY, GZIP, LZO, ZSTD, LZ4_RAW.

As expected, I will use Carpet with the default configuration that parquet-java brings, and the default compression level of each algorithm.

You can find the source code on GitHub, and the tests were done on a laptop with an AMD Ryzen 7 4800HS CPU and JDK 17.

File Size

To understand how each compression performs, we will take the equivalent CSV file as a reference.

Format	gov.it	NYC Taxis
CSV	1761 MB	2983 MB
UNCOMPRESSED	564 MB	760 MB
SNAPPY	220 MB	542 MB
GZIP	146 MB	448 MB
ZSTD	148 MB	430 MB
LZ4_RAW	209 MB	547 MB
LZO	215 MB	518 MB

In both tests, compression with GZip and Zstandard stands out as the most efficient.

Using only Parquet encoding techniques, the file size can be reduced to 25-32% of the original CSV size. Applying additional compression reduces it to between 9% and 15% of the CSV size.

Writing

How much overhead does compressing the information bring?

If we write the same information three times and average the seconds, we get:

Algorithm	gov.it	NYC Taxis
UNCOMPRESSED	25.0	57.9
SNAPPY	25.2	56.4
GZIP	39.3	91.1
ZSTD	27.3	64.1
LZ4_RAW	24.9	56.5
LZO	26.0	56.1

SNAPPY, LZ4, and LZO achieve similar times to not compressing, while ZSTD adds a bit of overhead. GZIP performs the worst, worsening the writing time by 50%.

Reading

Reading the files is faster than writing since fewer computations are needed.

Reading all the columns from the file, the times in seconds are:

Algorithm	gov.it	NYC Taxis
UNCOMPRESSED	11.4	37.4
SNAPPY	12.5	39.9
GZIP	13.6	40.9
ZSTD	13.1	41.5
LZ4_RAW	12.8	41.6
LZO	13.1	41.1

Reading times are close to not compressing the information, and the overhead of decompression is between 10% and 20%.

Conclusion

No algorithm has stood out significantly over the others in reading and writing times, all being within a similar range. In most cases, compressing the information compensates for the space savings (and transmission) over the time penalty.

In these two use cases, the determining factor for selecting one or another would probably be the compression ratio achieved, with ZSTD and Gzip standing out (but with poor writing time).

Each algorithm has its strengths, so the best option is to test with your data, considering which factor is more important:

Minimizing storage usage, because you store a lot of data that you rarely use.
Minimizing file generation time.
Minimizing reading time, since files are read many times.

As with everything in life, it's a trade-off, and you will have to see what compensates the most. In Carpet, by default, if you configure nothing, it compresses with Snappy.

Implementation Details

The value must be one of those available in the CompressionCodecName enum. Associated with each enum value is the name of the class implementing the algorithm:

public enum CompressionCodecName {
  UNCOMPRESSED(null, CompressionCodec.UNCOMPRESSED, ""),
  SNAPPY("org.apache.parquet.hadoop.codec.SnappyCodec", CompressionCodec.SNAPPY, ".snappy"),
  GZIP("org.apache.hadoop.io.compress.GzipCodec", CompressionCodec.GZIP, ".gz"),
  LZO("com.hadoop.compression.lzo.LzoCodec", CompressionCodec.LZO, ".lzo"),
  BROTLI("org.apache.hadoop.io.compress.BrotliCodec", CompressionCodec.BROTLI, ".br"),
  LZ4("org.apache.hadoop.io.compress.Lz4Codec", CompressionCodec.LZ4, ".lz4hadoop"),
  ZSTD("org.apache.parquet.hadoop.codec.ZstandardCodec", CompressionCodec.ZSTD, ".zstd"),
  LZ4_RAW("org.apache.parquet.hadoop.codec.Lz4RawCodec", CompressionCodec.LZ4_RAW, ".lz4raw");
  ...

Parquet will use reflection to instantiate the specified class, which must implement the CompressionCodec interface. If you look at its source code, you will see that it is within the Hadoop project, not Parquet. This shows how coupled Parquet is with Hadoop in the Java implementation.

To use one of the codecs, you must ensure you have added a JAR containing its implementation as a dependency.

Not all implementations are available in the transitive dependencies you have when adding parquet-java, or you may have excluded Hadoop dependencies too aggressively.

In the org.apache.parquet:parquet-hadoop dependency, the implementations of SnappyCodec, ZstandardCodec, and Lz4RawCodec are included, which transitively imports the snappy-java, zstd-jni, and aircompressor dependencies with the actual implementations of the three algorithms.

In the hadoop-common:hadoop-common dependency, the implementation of GzipCodec is included.

Where are the implementations of BrotliCodec and LzoCodec? They are not in any of the Parquet or Hadoop dependencies, so if you use them without adding additional dependencies, your application will not work with files compressed with those formats.

To support LZO, you need to add the dependency org.anarres.lzo:lzo-hadoop to your pom or gradle files.
Even more complex is the case of Brotli: the dependency is not in Maven Central, and you must also add the JitPack repository.

Working with Parquet files in Java using Carpet

Jerónimo López — Wed, 19 Jun 2024 16:55:22 +0000

After some time working with Parquet files in Java using the Parquet Avro library, and studying how it worked, I concluded that despite being very useful in multiple use cases and having great potential, the documentation and ecosystem needed for adoption in the Java world was very poor.

Many people are using suboptimal solutions (CSV or JSON files), applying more complex solutions (Spark), or using languages they are not familiar with (Python) because they don't know how to work with Parquet files easily. That's why I decided to write this series of articles.

Once you understand it and have the examples, everything is easier. But, can it be even easier? Can we avoid the hassle of using strange libraries that serialize other formats? Yes, it should be even easier.

That's why I decided to implement an Open Source library that makes working with Parquet from Java extremely simple, something that covers it: Carpet.

Carpet is a Java library that serializes and deserializes Parquet files to Java 17 Records, abstracting you (if you want) from the particularities of Parquet and Hadoop, and minimizing the number of necessary dependencies, because it works directly with Parquet code. It is available on Maven Central and you can find its source code on GitHub.

Hello world

Carpet works by reflection: it inspects your class model and there is no need to define an IDL, implement interfaces, or use annotations. Carpet is based on Java records, the primitive created by the JDK for Data Oriented Programming.

Continuing with the same examples from previous articles, we will have a collection of Organization objects, which have a list of Attributes:

record Org(String name, String category, String country, Type type, List<Attr> attributes) { }

record Attr(String id, byte quantity, byte amount, boolean active, double percent, short size) { }

enum Type { FOO, BAR, BAZ }

With Carpet, it is not necessary to create special classes or perform transformations. Carpet works directly with your model, as long as it fits the Parquet schema you need.

Serialization

With Carpet, you don't need to use Parquet writers or Hadoop classes:

try (OutputStream outputStream = new FileOutputStream(filePath)) {
    try (CarpetWriter writer = new CarpetWriter<>(outputStream, Org.class)) {
        writer.write(organizations);
    }
}

The code can be found on GitHub.

If your records match the required Parquet schema, class conversion is not necessary. If you don't need special Parquet configuration, you don't have to create builders, and you can use a Java OutputStream directly.

By reflection, it creates the Parquet schema, using the names and types of the fields in your records as column names and types.

Carpet supports complex data structures, as long as all objects are records, collections (List, Set, etc.), and maps.

Deserialization

Deserialization is equally simple, or even simpler.

List<Org> organizations = new CarpetReader<>(new File(filePath), Org.class).toList();

You can also iterate through the file with a stream:

List<Org> organizations = new CarpetReader<>(new File(filePath), Org.class).stream()
    .filter(this::somePredicate)
    .toList();

The code can be found on GitHub.

Since Carpet uses reflection, it conventionally expects the types and names of the fields to match those of the columns in the Parquet file.

None of the Parquet or Hadoop classes are imported into your code.

Deserialization using a projection

Carpet reads only the columns that are defined in the records and ignores any other columns that exist in the file. Defining a projection with a subset of attributes is as simple as defining a record in Java:

record OrgProjection(String name, String category, String country, Type type) { }

var organizations = new CarpetReader<>(new File(filePath), OrgProjection.class).toList();

In this case, reading time is reduced to hundreds of milliseconds.

The code can be found on GitHub.

The Parquet way

If for any reason you need to customize some parameter of file generation or use it with Hadoop, Carpet provides an implementation of the ParquetWriter and ParquetReader builders. This way, all Parquet configurations are exposed.

Serialization

We will need to instantiate a Parquet writer:

OutputFile outputFile = new FileSystemOutputFile(new File(filePath));
try (ParquetWriter.<Org> writer = CarpetParquetWriter.<Org>builder(outputFile, Org.class)
        .withCompressionCodec(CompressionCodecName.GZIP)
        .withWriteMode(Mode.OVERWRITE)
        .build()) {
    for (Org org : organizations) {
        writer.write(org);
    }
}

The code can be found on GitHub.

Carpet implements a ParquetWriter<T> builder with all the logic to convert Java records to Parquet API calls.

To avoid using Hadoop classes (and importing all their dependencies), Carpet implements the InputFile and OutputFile interfaces using regular files.

Therefore:

OutputFile and ParquetWriter are classes defined by the Parquet API
CarpetParquetWriter and FileSystemOutputFile are classes implemented by Carpet
Org and Attr are Java records from your domain, unrelated to Parquet or Carpet

Carpet implicitly generates the Parquet schema from the fields of your records.

Deserialization

We will need to instantiate a Parquet reader using the CarpetParquetReader builder:

InputFile inputFile = new FileSystemInputFile(new File(filePath));
try (ParquetReader<Org> reader = CarpetParquetReader.builder(inputFile, Org.class).build()) {
    List<Org> organizations = new ArrayList<>();
    Org next = null;
    while ((next = reader.read()) != null) {
        organizations.add(next);
    }
    return organizations;
}

You can find the code on GitHub.

Parquet defines a class called ParquetReader<T>, and Carpet implements it with CarpetParquetReader, handling the logic to convert internal data structures of Parquet to your Java records.

In this case:

InputFile and ParquetReader are classes defined by the Parquet API
CarpetParquetReader and FileSystemOutputFile are classes implemented by Carpet
Org (and Attr) are Java records from your domain, unrelated to Parquet

The instantiation of the ParquetReader class is also done with a Builder to maintain the pattern followed by Parquet.

Carpet validates that the schema of the Parquet file is compatible with the Java records. If not, it throws an exception.

Performance

With identical schemas and data, the file sizes compared to parquet-avro and parquet-protobuf are the same. However, what is the overhead cost of using reflection?

Library	Serialization	Deserialization
Parquet Avro	15,381 ms	7,665 ms
Parquet Protocol Buffers	16,174 ms	11,025 ms
Carpet	12,769 ms	8,881 ms

Writing, Carpet is 20% faster than using Avro and Protocol Buffers. The overhead of reflection is less than the work required to create Avro or Protocol Buffers objects.

In terms of reading, Carpet is slightly slower than the fastest version of Parquet Avro. The use of reflection does not significantly penalize performance, and in return, we avoid using custom data types of the library.

Conclusion

Parquet is a very powerful format, yet underutilized in the Java ecosystem. This is partly due to lack of awareness and the difficulty in working with it, and partly because being a binary format, it is not very comfortable to work with it.

Even if you're not into Big Data, Parquet can still be useful in scenarios involving large datasets. Often, due to unfamiliarity, complex or inefficient solutions and architectures are adopted.

The format, with its schema, ensures that the defined types are satisfied or the data cannot be null. How many times have you struggled parsing a CSV file?

Carpet provides a very simple API, making it extremely easy to write and process Parquet files in 99% of use cases. For me, working with Parquet files is now more convenient than CSVs.

Carpet is an open-source library under the Apache 2.0 license. You can find its source code on GitHub and it's available on Maven Central.

The README.md of the project provides a detailed explanation of its various functionalities, customization options, and how to use its API. I encourage you to use Carpet and share your feedback or tell me about your use cases working with Parquet.

Working with Parquet files in Java using Protocol Buffers

Jerónimo López — Thu, 07 Dec 2023 05:00:00 +0000

This post continues the series of articles about working with Parquet files in Java. This time, I'll explain how to do it using the Protocol Buffers (PB) library.

Finding examples and documentation on how to use Parquet with Avro is challenging, but with Protocol Buffers, it's even more complicated.

Protocol Buffers and Parquet both support complex data structures, offering a direct and clear mapping between them.

I will base this on the same example I used in previous articles talking about serialization. The code will be very similar to that in the article about Protocol Buffers.

In the example, we will work with a collection of Organization objects, which in turn have a list of Attributes:

record Org(String name, String category, String country, Type type, List<Attr> attributes) {
}

record Attr(String id, byte quantity, byte amount, boolean active, double percent, short size) {
}

enum Type {
  FOO, BAR, BAZ
}

In Parquet with PB we must also use classes generated from Protocol Buffers IDL. This capability is specific to PB, not Parquet, but it is inherited by parquet-protobuf, the library that implements this integration.

Internally, the library transforms the PB schema into the Parquet schema, so most tools and libraries that can work with PB classes will be able to work indirectly with Parquet with few changes.

Apart from the classes used to write or read the files, the remaining logic for building the objects generated by PB or reading their data remains unchanged compared to PB serialization.

Serialization

We will need to instantiate a Parquet writer that supports the writing of objects created by PB:

Path path = new Path("/tmp/my_output_file.parquet");
OutputFile outputFile = HadoopOutputFile.fromPath(path, new Configuration());
ParquetWriter<Organization> writer = ProtoParquetWriter.<Organization>builder(outputFile)
      .withMessage(Organization.class)
      .withWriteMode(Mode.OVERWRITE)
      .config(ProtoWriteSupport.PB_SPECS_COMPLIANT_WRITE, "true")
      .build();

Parquet defines a class named ParquetWriter<T> and the parquet-protobuf library extends it by implementing in ProtoParquetWriter<T> the logic of converting PB objects into calls to the Parquet API. The object we will serialize is Organization, which has been generated using the PB utility and implements the PB API.

The Path class is not the one from java.nio.file, but a Hadoop-specific abstraction for referencing file paths. Whereas the OutputFile class is Parquet's file abstraction with the capability to write to them.

Therefore:

Path, OutputFile, HadoopOutputFile, and ParquetWriter are classes defined by the Parquet API
ProtoParquetWriter is a class defined by the parquet-protobuf API, a library that encapsulates Parquet with Protocol Buffers
Organization and Attribute are classes generated by the PB utility, not related to Parquet

The way to create an instance of ParquetWriter is through a Builder, where you can configure many Parquet-specific parameters or those of the library we are using (Protocol Buffers). For example:

withMessage: class generated with Protocol Buffers that we want to persist (which internally defines the schema)
withCompressionCodec: compression method to use: SNAPPY, GZIP, LZ4, etc. By default, it doesn't configure any.
withWriteMode: by default it is CREATE, so if the file already existed, it would not overwrite it and would throw an exception. To avoid this, use OVERWRITE.
withValidation: if we want it to validate the data types passed against the defined schema.

Generic configuration can be passed using the config(String property, String value) method. In this case, we configure it to internally use a three-level structure to represent nested lists.

Once the ParquetWriter class is instantiated, the major complexity lies in transforming your POJOs into the Organization classes generated from the PB IDL:

Path path = new Path("/tmp/my_output_file.parquet");
OutputFile outputFile = HadoopOutputFile.fromPath(path, new Configuration());
try (ParquetWriter<Organization> writer = ProtoParquetWriter.<Organization>builder(outputFile)
      .withMessage(Organization.class)
      .withWriteMode(Mode.OVERWRITE)
      .config(ProtoWriteSupport.PB_SPECS_COMPLIANT_WRITE, "true")
      .build()) {
    for (Org org : organizations) {
        var organizationBuilder = Organization.newBuilder()
            .setName(org.name())
            .setCategory(org.category())
            .setCountry(org.country())
            .setType(OrganizationType.forNumber(org.type().ordinal()));
        for (Attr attr : org.attributes()) {
            var attribute = Attribute.newBuilder()
                .setId(attr.id())
                .setQuantity(attr.quantity())
                .setAmount(attr.amount())
                .setActive(attr.active())
                .setPercent(attr.percent())
                .setSize(attr.size())
                .build();
            organizationBuilder.addAttributes(attribute);
        }
        writer.write(organizationBuilder.build());
    }
}

Instead of converting the entire collection of organizations and then writing it, we can convert and persist each Organization one by one.

You can find the code on GitHub.

Deserialization

Deserialization is very straightforward if we are willing to work with the classes generated by Protocol Buffers.

To read the file, we will need to instantiate a Parquet reader:

Path path = new Path(filePath);
InputFile inputFile = HadoopInputFile.fromPath(path, new Configuration());
ParquetReader<Organization.Builder> reader =
    ProtoParquetReader.<Organization.Builder>builder(inputFile).build()

Parquet defines a class named ParquetReader<T> and the parquet-protobuf library extends it by implementing in ProtoParquetReader the logic of converting Parquet's internal data structures into classes generated by Protocol Buffers.

InputFile is Parquet's file abstraction with the capability to read from them.

Therefore:

Path, InputFile, HadoopInputFile, and ParquetReader are classes defined by the Parquet API
ProtoParquetReader implements ParquetReader and is defined in parquet-protobuf, a library that encapsulates Parquet with PB
Organization (and Attribute) are classes generated by the PB utility, not related to Parquet

The instantiation of the ParquetReader class is also done with a Builder, although the options to configure are much fewer, as all its configuration is determined by the file we are going to read. The reader does not need to know if the file uses dictionary encoding or if it is compressed, so it is not necessary to configure it; it discovers this by reading the file.

It's important to note that the data type the reader will return is a Builder for Organization, rather than the Organization itself, and we will need to call the build() method:

Path path = new Path(filePath);
InputFile inputFile = HadoopInputFile.fromPath(path, new Configuration());
try (ParquetReader<Organization.Builder> reader = ProtoParquetReader.<Organization.Builder>builder(inputFile).build()) {
    List<Organization> organizations = new ArrayList<>();
    Organization.Builder next = null;
    while ((next = reader.read()) != null) {
        organizations.add(next.build());
    }
    return organizations;
}

If the IDL used to generate the code contained only a subset of the columns existing in the file, when reading it, it would be ignoring all the columns not present in the IDL. This would save us disk reads and data deserialization/decoding.

You can find the code on GitHub.

Performance

What performance does Parquet with Protocol Buffers offer when serializing and deserializing large volumes of data? To what extent do different compression options affect it? Should we choose compression with Snappy or no compression at all? And what about enabling the dictionary or not?

Leveraging the analyses I conducted previously on different serialization formats, we can get an idea of their strengths and weaknesses. The benchmarks were conducted on the same computer, making them comparable.

File Size

	Uncompressed	Snappy
Dictionary False	1 034 MB	508 MB
Dictionary True	289 MB	281 MB

Given the difference in sizes, we can see that in my synthetic example, the use of dictionaries compresses the information quite effectively, almost better than the Snappy algorithm itself. The decision to enable compression or not will depend on the performance penalty it entails.

Serialization Time

	Uncompressed	Snappy
Dictionary False	14,802 ms	15,450 ms
Dictionary True	16,018 ms	16,174 ms

The time is very similar in all cases, and we can say that the different compression techniques do not significantly affect the time taken.

Compared with other serialization formats, it takes between 50% (Jackson) and 300% (Protocol Buffers/Avro) longer, but in return, it achieves files that are between 60% (Protocol Buffers/Avro) and 90% (Jackson) smaller.

Deserialization Time

	Uncompressed	Snappy
Dictionary False	12 ,64 ms	13,028 ms
Dictionary True	10,492 ms	11,025 ms

In this case, the use of the dictionary has a significant impact on time, saving the effort of decoding information that is repeated. There is definitely no reason to disable this feature.

Compared to other formats, it is 100% slower than pure Protocol Buffers and 30% slower than pure Avro, but it is almost twice as fast as Jackson.

To put the performance into perspective, on my laptop, it reads 45,000 Organizations per second, which in turn contain almost 2.6 million instances of the Attribute type.

Conclusion

If you were already familiar with Protocol Buffers, most of the code and particularities related to PB will be recognizable to you. If not, it increases the entry barrier, as you need to learn about two distinct but similar technologies, and it's not always clear which part belongs to each one.

The biggest change compared to pure Protocol Buffers is the way to create the writer and reader objects, where we have to deal with all the configuration and particularities specific to Parquet.

Although possible, Protocol Buffers is not intended for serializing large amounts of data, so it's not comparable to Parquet using Protocol Buffers. If I had to choose between using Parquet with PB or Parquet with Avro, I would probably opt for the Avro version, as Avro is often used in the data engineering world and you could benefit more from the experience.

The data I used in the example is random and results may vary depending on the characteristics of your data. Test with your data to draw conclusions.

In environments where you write once and read multiple times, the time spent on serialization should not be a determining factor. More important are the file size, transfer time, or processing speed (especially if you can filter columns by making projections).

Despite using different compression and encoding techniques, the processing speed of files is quite high. Coupled with its ability to work with a typed schema, this makes Parquet a data exchange format to consider in projects with a high volume of data.

Working with Parquet files in Java using Avro

Jerónimo López — Sun, 26 Nov 2023 11:09:55 +0000

In the previous article, I wrote an introduction to using Parquet files in Java, but I did not include any examples. In this article, I will explain how to do this using the Avro library.

Parquet with Avro is one of the most popular ways to work with Parquet files in Java due to its simplicity, flexibility, and because it is the library with the most examples.

Both Avro and Parquet allow complex data structures, and there is a mapping between the types of one and the other.

The post will use the same example I used in previous articles talking about serialization. The code will be very similar to the article about Avro. For specific details about Avro, I refer you to that article.

In the example, we will work with a collection of Organization objects (Org), which have also a list of Attributes (Attr):

record Org(String name, String category, String country, Type type, List<Attr> attributes) {
}

record Attr(String id, byte quantity, byte amount, boolean active, double percent, short size) {
}

enum Type {
  FOO, BAR, BAZ
}

Similar to saving files in Avro format, this version of Parquet with Avro allows writing files using classes generated from the IDL or the GenericRecord data structure. This capability is specific to Avro, not Parquet, but is inherited by parquet-avro, the library that implements this integration.

Internally, the library transforms the Avro schema into the Parquet schema, so most tools and libraries that know how to work with Avro classes will be able to work indirectly with Parquet with few changes.

Using code generation

The only difference when compared to serializing in Avro format lies in the class used for writing or reading files; otherwise, the logic for building the Avro-generated classes and reading their data remains unchanged.

Serialization

We will need to instantiate a Parquet writer that supports the writing of objects created by Avro:

Path path = new Path("/tmp/my_output_file.parquet");
OutputFile outputFile = HadoopOutputFile.fromPath(path, new Configuration());
ParquetWriter<Organization> writer = AvroParquetWriter.<Organization>builder(outputFile)
    .withSchema(new Organization().getSchema())
    .withWriteMode(Mode.OVERWRITE)
    .config(AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE, "false")
    .build();

Parquet defines a class called ParquetWriter<T> and the parquet-avro library extends it implementing in AvroParquetWriter<T> the logic of converting Avro objects into calls to the Parquet API. The object we will serialize is Organization, which has been generated using the Avro utility and implements the Avro API.

The Path class is not the one existing in java.nio.file, but a Hadoop-specific abstraction for referencing file paths. Whereas the OutputFile class is Parquet's file abstraction with the capability to write to them.

Therefore:

Path, OutputFile, HadoopOutputFile, and ParquetWriter are classes defined by the Parquet API.
AvroParquetWriter is a class defined by the parquet-avro API, a library that encapsulates Parquet with Avro.
Organization and Attribute are classes generated by the Avro utility, not related to Parquet.

The way to create an instance of ParquetWriter is through a Builder, where you can configure many Parquet-specific parameters or those of the library we are using (Avro). For example:

withSchema: schema of the Organization class in Avro, which internally converts to a Parquet schema.
withCompressionCodec: compression method to use: SNAPPY, GZIP, LZ4, etc. By default, it doesn't configure any.
withWriteMode: by default it is CREATE, so if the file already existed, it would not overwrite it and would throw an exception. To avoid this, use OVERWRITE.
withValidation: if we want it to validate the data types passed against the defined schema.
withBloomFilterEnabled: if we want to enable the creation of bloom filters.

A most generic configuration of both libraries (not defined in the API) can be passed with the config(String property, String value) method. In this case, we configure it to internally use a 3-level structure to represent nested lists.

Once the ParquetWriter class is instantiated, the greatest complexity lies in transforming your POJOs into the Organization classes generated from Avro's IDL. The complete code would be as follows:

Path path = new Path("/tmp/my_output_file.parquet");
OutputFile outputFile = HadoopOutputFile.fromPath(path, new Configuration());
try (ParquetWriter<Organization> writer = AvroParquetWriter.<Organization>builder(outputFile)
    .withSchema(new Organization().getSchema())
    .withWriteMode(Mode.OVERWRITE)
    .config(AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE, "false")
    .build()) {
  for (var org : organizations) {
    List<Attribute> attrs = org.attributes().stream()
      .map(a -> Attribute.newBuilder()
        .setId(a.id())
        .setQuantity(a.quantity())
        .setAmount(a.amount())
        .setSize(a.size())
        .setPercent(a.percent())
        .setActive(a.active())
        .build())
      .toList();
    Organization organization = Organization.newBuilder()
      .setName(org.name())
      .setCategory(org.category())
      .setCountry(org.country())
      .setOrganizationType(OrganizationType.valueOf(org.type().name()))
      .setAttributes(attrs)
      .build();
    writer.write(organization);
  }
}

Instead of converting the entire collection of organizations and then writing it, we can convert and persist each Organization one by one.

You can find the code on GitHub.

Deserialization

Deserialization is very straightforward if we agree to work with the classes generated by Avro.

To read the file, we will need to instantiate a Parquet reader:

Path path = new Path(filePath);
InputFile inputFile = HadoopInputFile.fromPath(path, new Configuration());
ParquetReader<Organization> reader = AvroParquetReader.<Organization>builder(inputFile).build();

Parquet defines a class called ParquetReader<T> and the parquet-avro library extends it by implementing in AvroParquetReader the logic of converting Parquet's internal data structures into the classes generated by Avro.

In this case, InputFile is Parquet's file abstraction with the capability to read from them.

Therefore:

Path, InputFile, HadoopInputFile, and ParquetReader are classes defined by the Parquet API.
AvroParquetReader implements ParquetReader and is defined in parquet-avro, a library that encapsulates Parquet with Avro.
Organization (and Attribute) are classes generated by the Avro utility, not related to Parquet.

Path path = new Path(filePath);
InputFile inputFile = HadoopInputFile.fromPath(path, new Configuration());
try (ParquetReader<Organization> reader = AvroParquetReader.<Organization>builder(inputFile).build()) {
    List<Organization> organizations = new ArrayList<>();
    Organization next = null;
    while ((next = reader.read()) != null) {
        organizations.add(next);
    }
    return organizations;
}

If the IDL used to generate the code contains a subset of the attributes persisted in the file, when reading it we would be ignoring all the columns not present in the IDL. This would save us from disk reads and the deserialization/decoding of data.

You can find the code on GitHub.

Using GenericRecord

Here, it will not be necessary to generate any code, and we will work with the GenericRecord class provided by Avro, but the code will be a bit more verbose.

Serialization

As we do not have generated files containing the embedded schema, we need to programmatically define the Avro schema we are going to use. The code is the same as in the article about Avro:

Schema attrSchema = SchemaBuilder.record("Attribute")
  .fields()
  .requiredString("id")
  .requiredInt("quantity")
  .requiredInt("amount")
  .requiredInt("size")
  .requiredDouble("percent")
  .requiredBoolean("active")
  .endRecord();
var enumSymbols = Stream.of(Type.values()).map(Type::name).toArray(String[]::new);
Schema orgsSchema = SchemaBuilder.record("Organizations")
  .fields()
  .requiredString("name")
  .requiredString("category")
  .requiredString("country")
  .name("organizationType").type().enumeration("organizationType")
  .symbols(enumSymbols).noDefault()
  .name("attributes").type().array().items(attrSchema).noDefault()
  .endRecord();
var typeField = orgsSchema.getField("organizationType").schema();
EnumMap<Type, EnumSymbol> enums = new EnumMap<>(Type.class);
enums.put(Type.BAR, new EnumSymbol(typeField, Type.BAR));
enums.put(Type.BAZ, new EnumSymbol(typeField, Type.BAZ));
enums.put(Type.FOO, new EnumSymbol(typeField, Type.FOO));

Instead of using an AvroParquetWriter of the type Organization, we create one of the type GenericRecord and construct instances of it as if it were a Map:

Path path = new Path(filePath);
OutputFile outputFile = HadoopOutputFile.fromPath(path, new Configuration());
try (ParquetWriter<GenericRecord> writer = AvroParquetWriter.<GenericRecord>builder(outputFile)
    .withSchema(orgsSchema)
    .withWriteMode(Mode.OVERWRITE)
    .config(AvroWriteSupport.WRITE_OLD_LIST_STRUCTURE, "false")
    .build()) {
  for (var org : organizations) {
    List<GenericRecord> attrs = new ArrayList<>();
    for (var attr : org.attributes()) {
      GenericRecord attrRecord = new GenericData.Record(attrSchema);
      attrRecord.put("id", attr.id());
      attrRecord.put("quantity", attr.quantity());
      attrRecord.put("amount", attr.amount());
      attrRecord.put("size", attr.size());
      attrRecord.put("percent", attr.percent());
      attrRecord.put("active", attr.active());
      attrs.add(attrRecord);
    }
    GenericRecord orgRecord = new GenericData.Record(orgsSchema);
    orgRecord.put("name", org.name());
    orgRecord.put("category", org.category());
    orgRecord.put("country", org.country());
    orgRecord.put("organizationType", enums.get(org.type()));
    orgRecord.put("attributes", attrs);
    writer.write(orgRecord);
  }
}

You can find the code on GitHub.

Deserialization

As in the original version of Avro, most of the work consists in converting the GenericRecord into our data structure. Because it behaves like a Map, we will have to cast the types of the values:

Path path = new Path(filePath);
InputFile inputFile = HadoopInputFile.fromPath(path, new Configuration());
try (ParquetReader<GenericRecord> reader = AvroParquetReader.<GenericRecord>builder(inputFile).build()) {
  List<Org> organizations = new ArrayList<>();
  GenericRecord record = null;
  while ((record = reader.read()) != null) {
    List<GenericRecord> attrsRecords = (List<GenericRecord>) record.get("attributes");
    var attrs = attrsRecords.stream().map(attr -> new Attr(attr.get("id").toString(),
        ((Integer) attr.get("quantity")).byteValue(),
        ((Integer) attr.get("amount")).byteValue(),
        (boolean) attr.get("active"),
        (double) attr.get("percent"),
        ((Integer) attr.get("size")).shortValue())).toList();
    Utf8 name = (Utf8) record.get("name");
    Utf8 category = (Utf8) record.get("category");
    Utf8 country = (Utf8) record.get("country");
    Type type = Type.valueOf(record.get("organizationType").toString());
    organizations.add(new Org(name.toString(), category.toString(), country.toString(), type, attrs));
  }
  return organizations;
}

As we are using the Avro interface, it maintains its logic that Strings are encoded within the Utf8 class and it will be necessary to extract their values.

You can find the code on GitHub.

By default, when reading the file, it deserializes all fields of each row because it does not know the schema of what you need to read, and processes everything. If you want a projection of the fields, you must pass it in the form of an Avro schema when creating the ParquetReader:

Schema projection = SchemaBuilder.record("Organizations")
  .fields()
  .requiredString("name")
  .requiredString("category")
  .requiredString("country")
  .endRecord();
Configuration configuration = new Configuration();
configuration.set(AvroReadSupport.AVRO_REQUESTED_PROJECTION, orgsSchema.toString());
try (ParquetReader<GenericRecord> reader = AvroParquetReader.<GenericRecord>builder(inputFile)
  .withConf(configuration)
  .build()) {
....

The rest of the process would be the same, but with fewer fields. You can see the entire source code of the example here.

Performance

What performance does Parquet Avro offer when serializing and deserializing a large volume of data? To what extent do the different compression options influence? Do we choose compression with Snappy or no compression at all? And what about activating the dictionary or not?

Taking advantage of the analyses I did previously on different serialization formats, we can get an idea of their strengths and weaknesses. The benchmarks were done on the same computer, so they are comparable to give us an idea.

File Size

Both using code generation and GenericRecord, the result is the same, as they are different ways of defining the same schema and persisting the same data:

	Uncompressed	Snappy
Dictionay False	1,034 MB	508 MB
Dictionay True	289 MB	281 MB

Given the difference in sizes, we can see that in my synthetic example, the use of dictionaries compresses the information significantly, even more than the Snappy algorithm itself. The decision to activate compression or not will depend on the performance penalty it entails.

Serialization Time

Using code generation:

	Uncompressed	Snappy
Dictionary False	14,386 ms	14,920 ms
Dictionary True	15,110 ms	15,381 ms

Using GenericRecord:

	Uncompressed	Snappy
Dictionary False	15,287 ms	15,809 ms
Dictionary True	16,119 ms	16,432 ms

The time is very similar in all cases, and we can say that the different compression techniques do not significantly affect the time spent.

There are no notable time differences between generated code and the use of GenericRecord, so performance should not be a determining factor in choosing a solution.

Compared to other serialization formats, it takes between 40% (Jackson) and 300% (Protocol Buffers/Avro) more time, but in return achieves files that are between 70% (Protocol Buffers/Avro) and 90% (Jackson) smaller.

Deserialization Time

Using code generation:

	Uncompressed	Snappy
Dictionary False	10,722 ms	10,736 ms
Dictionary True	7,707 ms	7,665 ms

Using GenericRecord:

	Uncompressed	Snappy
Dictionary False	12,089 ms	11,931 ms
Dictionary True	8,374 ms	8,451 ms

In this case, the use of the dictionary has a significant impact on time, as it saves decoding information that is repeated. There is definitely no reason to disable this functionality.

If we compare with other formats, it is twice as slow as Protocol Buffers and on par with Avro, but more than twice as fast as Jackson.

To put the performance into perspective, on my laptop, it reads 50,000 Organizations per second, which in turn contain almost 3 million instances of type Attribute, per second!

Deserialization Time Using a Projection

What is the performance like if we use a projection and we only read three fields of the Organization object and ignore its collection of attributes?

	Uncompressed	Snappy
Dictionay False	289 ms	304 ms
Dictionay True	195 ms	203 ms

We confirm the promise that if we access a subset of columns, we will read and decode much less information. In this case, it takes only 2.5% of the time, or in other words, it is 40 times faster at processing the same file.

This is where Parquet shows its full power, by allowing to read and decode a subset of data quickly, taking advantage of how the data is distributed in the file.

Conclusion

If you are already using Avro or are familiar with it, most of the code and nuances related to Avro will be familiar to you. If you are not, it increases the entry barrier, as you have to learn about two different technologies, and it may not be clear what corresponds to each.

The major change compared to using only Avro is how the writer and reader objects are created, where we will have to deal with all the configuration and particularities specific to Parquet.

If I had to choose between using only Avro or Parquet with Avro, I would choose the latter, as it produces more compact files and we have the opportunity to take advantage of the columnar format.

The data I have used in the example are synthetic, and the results may vary depending on the characteristics of your data. I recommend doing tests, but unless all your values are very random, the compression rates will be high.

In environments where you write once and read multiple times, the time spent serializing should not be decisive. More important, for example, are the consumption of your storage, the file transfer time, or the processing speed (especially if you can filter the columns you access).

Despite using different compression and encoding techniques, the file processing time is quite fast. Along with its ability to work with a typed schema, this makes it a data interchange format to be considered in projects with a heavy load of data.

Working with Parquet files in Java

Jerónimo López — Tue, 21 Nov 2023 06:00:00 +0000

Parquet is a widely used format in the Data Engineering realm and holds significant potential for traditional Backend applications. This article serves as an introduction to the format, including some of the unique challenges I've faced while using it, to spare you from similar experiences.

Introduction

Apache Parquet, released by Twitter and Cloudera in 2013, is an efficient and general-purpose columnar file format for the Apache Hadoop ecosystem. Inspired by Google's paper "Dremel: Interactive Analysis of Web-Scale Datasets", Parquet is optimized to support complex and nested data structures.

Although it emerged almost simultaneously with ORC from Hortonworks and Facebook, it appears that Parquet has become the most used format.

Unlike row-oriented formats, Parquet organizes data by columns, enabling more efficient data persistence through advanced encoding and compression techniques.

Logical table

Row Oriented storage:

Column Oriented Storage:

When it comes to reading time, Parquet's columnar data organization proves beneficial. If there's no need to access every column, this arrangement allows you to bypass reading and processing numerous data blocks, significantly enhancing efficiency.

// Detect dark theme var iframe = document.getElementById('tweet-1725849122823983419-35'); if (document.body.className.includes('dark-theme')) { iframe.src = "https://platform.twitter.com/embed/Tweet.html?id=1725849122823983419&theme=dark" }

Columnar data formats are frequently employed in analytical database systems (such as Cassandra, BigQuery, ClickHouse, and QuestDB) and in large dataset processing systems like Apache Arrow and ORC.

Format

Data in Parquet is stored in binary form, making it unreadable when printed on the console, unlike other text-based formats like JSON, XML, or CSV. The binary storage format of Parquet offers benefits in terms of efficiency and performance, but lacks the human-readable aspect inherent to text-based data formats.

Given that Parquet is designed to handle vast volumes of data, it is practical to include the schema information directly within the files, as well as additional statistical metadata. This feature is particularly advantageous for working with files when their schema is not previously known. Each Parquet file inherently contains all the necessary details to deduce its schema, enabling seamless data reading and analysis, even when dealing with unknown or variable data structures.

Parquet supports basic data types, with the ability to extend them through logical types, giving them their own semantics:

BOOLEAN: 1-bit boolean (boolean in Java)
INT32: 32-bit signed integer (int in Java)
INT64: 64-bit signed integer (long in Java)
INT96: 96-bit signed integer (no direct equivalent in Java)
FLOAT: 32-bit IEEE floating-point value (float in Java)
DOUBLE: 64-bit IEEE floating-point value (double in Java)
BYTE_ARRAY: byte array of indeterminate size
FIXED_LEN_BYTE_ARRAY: fixed-size byte array

More complex data types such as Strings, Enums, UUIDs, and different kinds of date formats can be efficiently constructed using these foundational types. This ability demonstrates the flexibility of Parquet in accommodating a wide range of data structures by building upon basic data types.

The format supports persisting complex data structures, lists, and maps in a nested manner, which opens the door to storing any type of data that can be structured as a Document.

Historically, collections can be represented internally in multiple ways, depending on how you want to handle the possibility of a collection being null or empty. Additionally, as an implementation detail, how to name each element of the collection can vary between implementations of the format. These variations have led to different utilities in various languages generating files with differences that make them incompatible.

The official representation of collections in Parquet is now defined, yet many tools still generate files with legacy formats to preserve backward compatibility. This requires an explicit configuration to ensure files are written in the standardized format. For instance, when using Pandas with PyArrow, the use_compliant_nested_type option must be enabled, while in Java Avro Parquet, the WRITE_OLD_LIST_STRUCTURE_DEFAULT flag should be disabled.

While Parquet utilizes an IDL to define the data format within a file (its schema), it lacks a direct, standard tool for generating Java code from an IDL to facilitate the serialization and deserialization of Parquet data in simple Java classes.

Parquet inherently includes data compression within its encoding process. It automatically applies techniques like Run-length encoding (RLE) or Bit Packing. Additionally, it offers the option to compress data blocks with compressors such as Snappy, GZip, or LZ4, and to use Dictionaries for the normalization of repeated values.

Snappy compression is typically employed as the default, due to its favorable balance between compression efficiency and CPU time consumption.

When serializing or deserializing large amounts of data, Parquet allows us to write or read records one at a time, obviating the need for retaining all data in memory, unlike Protocol Buffers or FlatBuffers. You can serialize a stream of records or iterate through a file reading records one by one.

Documentation

Despite the format's significance in the world of Data Engineering, documentation on its basic use is quite scarce, particularly in the context of Java.

How would you feel if, to learn about how to read or write JSON files, you had to go through Pandas or Spark and it wasn't straightforward to do it directly? That's the sensation you get when you start studying Parquet.

High-level tools commonly used by Data Engineers, such as Pandas and Spark, offer built-in methods to seamlessly export and import data to Parquet (and other formats), simplifying the process by hiding the underlying complexities. However, finding documentation and examples for using Parquet independently of these tools is challenging, as the information is scattered across multiple articles written by different individuals.

What would you say if, to read or write JSON files, you had to go through other tools/formats such as Avro or Protocol Buffers, and there was no library that supported it directly? This is the scenario encountered with Parquet.

The absence of a straightforward library specifically for Parquet files, necessitating the use of third-party libraries for other format serializations, makes it more challenging to become proficient with Parquet.

Libraries

The Parquet library in Java does not offer a direct way to read or write Parquet files. Just as the Jackson library handles JSON files or the Protocol Buffers library works with its own format, Parquet does not include a function to read or write Java Objects (POJOs) or Parquet-specific data structures.

When working with Parquet in Java, there are two main approaches:

Using the low-level API provided by the Parquet library (this would be equivalent to processing the tokens of a JSON or XML parser).
Employing the functionalities of other serialization libraries, like Avro or Protocol Buffers.

Among the libraries that make up the Apache Parquet project in Java, there are specific libraries that use Protocol Buffers or Avro classes and interfaces for reading and writing Parquet files. These libraries employ the low-level API of parquet-mr to convert objects of Avro or Protocol Buffers type into Parquet files and vice versa.

To summarize, while working with Parquet in Java, you'll engage with three types of classes, each associated with three different APIs:

The API of your chosen serialization library, which provides the way to define serialized Objects and interact with them.
The API of the wrapper library for the serialization library you have chosen, with the readers and writers for these Objects.
The API of the parquet-mr low-level library itself that defines common interfaces and configurations, and handles the actual serialization process.

The most user-friendly and flexible library, often encountered in internet examples, is Avro. However, Protocol Buffers is also a viable option.

The use of utilities from these formats does not imply that the information is serialized twice, first through an intermediate format. Instead, it involves reusing the classes generated by Avro or Protocol Buffers, which hold the data to be stored. Each wrapper implementation uses the Parquet MR API.

Abstraction Over Files

The Parquet library is agnostic to the location of the data - it could be on a local file system, within a Hadoop cluster, or stored in S3.

To provide an abstraction layer for file locations, Parquet defines the interfaces org.apache.parquet.io.OutputFile and org.apache.parquet.io.InputFile. These interfaces contain methods to create specialized types of Output and Input Streams for data management.

For those interfaces, it provides an implementation responsible for implementing access to files in Hadoop, SFTP, or local files:

These implementations, in turn, require to reference files using the org.apache.hadoop.fs.Path class, which is unrelated to the Java Path class, along with an org.apache.hadoop.conf.Configuration.

To reference a file, we would need to write code like this:

Path path = new Path("/tmp/my_file.parquet");
OutputFile outputFile = HadoopOutputFile.fromPath(path, new Configuration());
InputFile inputFile = HadoopInputFile.fromPath(path, new Configuration());

But fortunately, this won't last long, as a second implementation (that allows working with local files only) has recently been merged to master, and it is being decoupled from the Hadoop configuration class. No release has yet included this, but it is expected to be available in version 1.14.0.

Dependencies

One of the major drawbacks of using Parquet in Java is the large number of transitive dependencies that its libraries have.

Parquet was conceived to be used in conjunction with Hadoop. The project on GitHub for the Java implementation is named parquet-mr, and mr stands for Map Reduce. As you can see, the package of the file classes refers to hadoop.

Over time, it has evolved and become more independent, but it has not been able to completely decouple from Hadoop and still has many transitive dependencies from the libraries used by Hadoop (ranging from a Jetty server to a Kerberos or Yarn clients).

If your project is going to use Hadoop, all those dependencies will be necessary, but if you intend to use standard files outside of Hadoop, it makes your application heavier. I suggest excluding those transitive dependencies in your pom.xml or build.gradle.

In addition to including a lot of unnecessary code, this can pose a problem when resolving conflicts with versions of transitive dependencies that you are also using.

If you don't exclude any dependencies, you might end up with more than 130 JARs and 75 MB in your deployable artifact. However, by selectively excluding unused dependencies, I've been able to trim this down to just 30 JARs, with a total size of 23 to 29MB.

As previously noted, efforts are underway to address this, but it's not all ready yet.

Conclusion

The Parquet format stands as a crucial tool in the Data Engineering ecosystem, providing an efficient solution for the storage and processing of large volumes of data.

Although its adoption has been solid in Big Data environments, its potential transcends this field and can also be leveraged in the world of traditional Backend.

Its column-oriented design, combined with advanced data compression and complex data structuring capabilities, make it a robust option for those seeking to enhance performance and efficiency in handling data within traditional Backend environments.

Despite its indisputable advantages, the adoption of Parquet as a data interchange format in Java application development faces obstacles, primarily due to the complexity of its low-level API and the lack of a high-level interface that simplifies its use. The need to rely on third-party libraries adds an additional layer of complexity and dependencies, and the scarcity of accessible documentation and concrete examples poses a significant barrier for many developers.

This post has been an introduction to the format, its advantages, and the WTFs I've encountered along the way; don't be discouraged. With this foundational knowledge, the forthcoming posts will focus on how to work with Parquet using different libraries:

Java serialization with Avro

Jerónimo López — Mon, 05 Dec 2022 14:23:58 +0000

In previous posts I've analyzed Protocol Buffers and FlatBuffers, using JSON as the baseline. In this post, I will analyze Apache Avro and compare it with the previously studied formats.

Apache Avro was developed as a component of the Apache Hadoop project and was released in 2009 under the Apache 2.0 license.

Avro should not be confused with Parquet. Often when searching for documentation on Parquet, you end up reading about Avro, confusing them. Although the Avro library is useful for generating Parquet files and both are widely used in the world of Big Data, the formats have no relation to each other.

As Protocol Buffers and Flat Buffers, you can predefine a schema (in JSON), and generates binary content that is not human-readable. It also has support for multiple languages.

It is not only used to persist large batches of data. For example, Avro is widely used in Kafka to serialize the information in its messages.

Because the format or schema can be included in the serialized data, code generation is optional, which makes it easy to build systems that process their files generically. It provides the flexibility of JSON, but with a more compact and efficient format.

I'm going to explore both approaches: code generation from the schema and generated programmatically.

IDL and code generation

The file with the equivalent schema to the example in the previous articles would be this one:

{
  "name": "Organization",
  "type": "record",
  "namespace": "com.jerolba.xbuffers.avro",
  "fields": [
    { "name": "name", "type": "string" }, 
    { "name": "category", "type": "string"}, 
    { "name": "organizationType",
      "type": {
        "type": "enum",
        "name": "OrganizationType",
        "symbols": ["FOO", "BAR", "BAZ"]
      }
    }, 
    { "name": "country", "type": "string" }, 
    { "name": "attributes",
      "type": {
        "type": "array",
        "items": {
          "type": "record",
          "name": "Attribute",
          "fields": [
            { "name": "id", "type": "string" }, 
            { "name": "quantity", "type": "int"}, 
            { "name": "amount", "type": "int"}, 
            { "name": "size", "type": "int"}, 
            { "name": "percent", "type": "double"}, 
            { "name": "active", "type": "boolean"}
          ]
        }
      }
    }
  ]
}

To generate all Java classes you need to download avro-tools,
and run it with parameters that reference where the IDL file is located and the target path of the generated files:

 java -jar avro-tools-1.11.1.jar compile schema ./src/main/resources/organizations.avsc ./src/main/java/

Or directly using Docker with an image ready to execute the command:

docker run --rm -v $(pwd)/src:/avro/src kpnnv/avro-tools:1.11.1 compile schema /avro/src/main/resources/organizations.avsc /avro/src/main/java/

Using generated classes

Serialization

Like Protocol Buffers, Avro does not directly serialize your POJOs, and you need to copy the information to the objects generated by the schema compiler.

But in this case, if you are persisting a collection of objects you don't need to have all them in memory because you can serialize objects one by one as you instantiate them.

The code needed to serialize the information starting from the POJOs would look like this:

DatumWriter<Organization> datumWriter = new SpecificDatumWriter<>(Organization.class);
var dataFileWriter = new DataFileWriter<>(datumWriter);
try (var os = new FileOutputStream("/tmp/organizations.avro")) {
  dataFileWriter.create(new Organization().getSchema(), os);
  for (var org : organizations) {
    List<Attribute> attrs = org.attributes().stream()
      .map(a -> Attribute.newBuilder()
        .setId(a.id())
        .setQuantity(a.quantity())
        .setAmount(a.amount())
        .setSize(a.size())
        .setPercent(a.percent())
        .setActive(a.active())
      .build())
    .toList();
    Organization organization = Organization.newBuilder()
      .setName(org.name())
      .setCategory(org.category())
      .setCountry(org.country())
      .setOrganizationType( OrganizationType.valueOf(org.type().name()))
      .setAttributes(attrs)
      .build();
    dataFileWriter.append(organization);
  }
  dataFileWriter.close();
}

Instead of converting the whole collection and then persisting it, we can convert and persist each Organization one by one.

Serialization time: 5,409 ms
File size: 846 MB
Compressed file size: 530 MB
Memory required: because we are directly serializing to OutputStream, it does not consume anything other than the necessary internal IO buffers (and original objects)
Library size (avro-1.11.1.jar + dependencies): 3,552,326 bytes
Size of generated classes: 37,283 bytes

Deserialization

Due to the internal representation of the data, Avro needs to move around the file parsing the data and you need to provide a seekable InputStream or directly a File. For example, you can not use directly an InputStream from an HTTP request.

With few lines you can read and process the whole object graph:

File file = new File("/tmp/organizations.avro");
DatumReader<Organization> datumReader = new SpecificDatumReader<>(Organization.class);
List<Organization> organizations = new ArrayList<>();
try (var dataFileReader = new DataFileReader<>(file, datumReader)) {
  while (dataFileReader.hasNext()) {
    organizations.add(dataFileReader.next());
  }
}

The objects are instances of the classes generated from the schema, not the original records. Because we are iterating a reader, we can transform each instance into our representation if necessary, without having both representations repeated in memory.

Deserialization time: 8,197 ms
Memory required: reconstructing all the object structures defined by the schema takes up 2,520 MB

Using Generic Record

In Avro, you can have the schema embedded in the binary file, and it allows you to read a serialized record without needing to know or agree on the schema in advance. This enables us to deserialize and know the contents of any file, without requiring code generation

You can define the schema at runtime and decide, based on your needs, the fields and structure of the serialized information.

Serialization

Instead of copying the data to the generated classes, you can do it through an Avro GenericRecord, which behaves like a Map. First, you need to define the Avro schema using code (or load it from a static JSON file):

Schema attrSchema = SchemaBuilder.record("Attribute")
    .fields()
    .requiredString("id")
    .requiredInt("quantity")
    .requiredInt("amount")
    .requiredInt("size")
    .requiredDouble("percent")
    .requiredBoolean("active")
    .endRecord();
var enumSymbols = Stream.of(Type.values()).map(Type::name)
    .toArray(String[]::new);
Schema orgsSchema = SchemaBuilder.record("Organizations")
    .fields()
    .requiredString("name")
    .requiredString("category")
    .requiredString("country")
    .name("organizationType").type().enumeration("organizationType")
                             .symbols(enumSymbols).noDefault()
    .name("attributes").type().array().items(attrSchema).noDefault()
    .endRecord();
//Auxiliar Map to encode Enums
var typeField = orgsSchema.getField("organizationType").schema();
EnumMap<Type, EnumSymbol> enums = new EnumMap<>(Type.class);
enums.put(Type.BAR, new EnumSymbol(typeField, Type.BAR));
enums.put(Type.BAZ, new EnumSymbol(typeField, Type.BAZ));
enums.put(Type.FOO, new EnumSymbol(typeField, Type.FOO));

And the code necessary to serialize the collection would look like this:

DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(orgsSchema);
var dataFileWriter = new DataFileWriter<>(datumWriter);
try (var os = new FileOutputStream("/tmp/organizations.avro")) {
  dataFileWriter.create(orgsSchema, os);
  for (var org : organizations) {
    List<GenericRecord> attrs = new ArrayList<>();
    for (var attr : org.attributes()) {
      GenericRecord attrRecord = new GenericData.Record(attrSchema);
      attrRecord.put("id", attr.id());
      attrRecord.put("quantity", attr.quantity());
      attrRecord.put("amount", attr.amount());
      attrRecord.put("size", attr.size());
      attrRecord.put("percent", attr.percent());
      attrRecord.put("active", attr.active());
      attrs.add(attrRecord);
    }
    GenericRecord orgRecord = new GenericData.Record(orgsSchema);
    orgRecord.put("name", org.name());
    orgRecord.put("category", org.category());
    orgRecord.put("country", org.country());
    orgRecord.put("organizationType", enums.get(org.type()));
    orgRecord.put("attributes", attrs);
    dataFileWriter.append(orgRecord);
  }
  dataFileWriter.close();
}

Because we are using the same Schema, and we are only changing how we serialize the data, the file size and memory used are the same.

The serialization time grows to 5,903 ms, 10% more than using generated code. The implementation of GenericRecord introduces a slight overhead.

Deserialization

Using a different Reader, the result of the deserialized data is again an untyped GenericRecord object. In this case, we need to convert each instance to the original data structure, mapping each type:

List<Org> organizations = new ArrayList<>();
DatumReader<GenericRecord> datumReader = new GenericDatumReader<>();
try (var dataFileReader = new DataFileReader<>(file, datumReader)) {
  while (dataFileReader.hasNext()) {
    GenericRecord record = dataFileReader.next();
    List<GenericRecord> attrsRecords = (List<GenericRecord>) record.get("attributes");
    var attrs = attrsRecords.stream().map(attr -> new Attr(attr.get("id").toString(),
      ((Integer) attr.get("quantity")).byteValue(),
      ((Integer) attr.get("amount")).byteValue(),
      (boolean) attr.get("active"),
      (double) attr.get("percent"),
      ((Integer) attr.get("size")).shortValue())).toList();
    Utf8 name = (Utf8) record.get("name");
    Utf8 category = (Utf8) record.get("category");
    Utf8 country = (Utf8) record.get("country");
    Type type = Type.valueOf(record.get("organizationType").toString());
    organizations.add(new Org(name.toString(), category.toString(), country.toString(), type, attrs));
  }
}

The code is verbose and is plenty of castings. Avro does not support the byte and short types, and they are persisted as int, so we need to downcast their values. As an optimization, Strings are created with an internal representation called Utf8.

The deserialization time grows to 8,471 ms, around 5% of overhead compared to the static code version.

Using optimized Generic Record

If you inspect the implementation of the GenericRecord class, you can see that get(String key) and put(String key, Object value) access by that key to a Map to get the index in an array. Since each attribute will always be in the same position when reading the file, we can access it only once and reuse its value with a variable, improving the execution time.

Serialization

Because you need to store the index of each field in the Schema, the code is even more verbose. After creating the Schema, the code is:

int idPos = attrSchema.getField("id").pos();
int quantityPos = attrSchema.getField("quantity").pos();
int amountPos = attrSchema.getField("amount").pos();
int activePos = attrSchema.getField("active").pos();
int percentPos = attrSchema.getField("percent").pos();
int sizePos = attrSchema.getField("size").pos();
int namePos = orgsSchema.getField("name").pos();
int categoryPos = orgsSchema.getField("category").pos();
int countryPos = orgsSchema.getField("country").pos();
int organizationTypePos = orgsSchema.getField("organizationType").pos();
int attributesPos = orgsSchema.getField("attributes").pos();

try (var os = new FileOutputStream("/tmp/organizations.avro")) {
  DatumWriter<GenericRecord> datumWriter = new GenericDatumWriter<>(orgsSchema);
  DataFileWriter<GenericRecord> dataFileWriter = new DataFileWriter<>(datumWriter);
  dataFileWriter.create(orgsSchema, os);
  for (var org : organizations) {
    List<GenericRecord> attrs = new ArrayList<>();
    for (var attr : org.attributes()) {
      GenericRecord attrRecord = new GenericData.Record(attrSchema);
      attrRecord.put(idPos, attr.id());
      attrRecord.put(quantityPos, attr.quantity());
      attrRecord.put(amountPos, attr.amount());
      attrRecord.put(sizePos, attr.size());
      attrRecord.put(percentPos, attr.percent());
      attrRecord.put(activePos, attr.active());
      attrs.add(attrRecord);
    }
    GenericRecord orgRecord = new GenericData.Record(orgsSchema);
    orgRecord.put(namePos, org.name());
    orgRecord.put(categoryPos, org.category());
    orgRecord.put(countryPos, org.country());
    orgRecord.put(organizationTypePos, enums.get(org.type()));
    orgRecord.put(attributesPos, attrs);
    dataFileWriter.append(orgRecord);
  }
  dataFileWriter.close();
}

The serialization time is 5,381 ms, very close to the time used in the version with generated code..

Deserialization

The code is very similar, we only add variables to know the position of each field:

List<Org> organizations = new ArrayList<>();
DatumReader<GenericRecord> datumReader = new GenericDatumReader<>();
try (DataFileReader<GenericRecord> dataFileReader = new DataFileReader<>(file, datumReader)) {
  Schema attributes = dataFileReader.getSchema().getField("attributes").schema().getElementType();
  int idPos = attributes.getField("id").pos();
  int quantityPos = attributes.getField("quantity").pos();
  int amountPos = attributes.getField("amount").pos();
  int activePos = attributes.getField("active").pos();
  int percentPos = attributes.getField("percent").pos();
  int sizePos = attributes.getField("size").pos();
  Schema orgs = dataFileReader.getSchema();
  int namePos = orgs.getField("name").pos();
  int categoryPos = orgs.getField("category").pos();
  int countryPos = orgs.getField("country").pos();
  int organizationTypePos = orgs.getField("organizationType").pos();
  while (dataFileReader.hasNext()) {
    GenericRecord record = dataFileReader.next();
    List<GenericRecord> attrsRecords = (List<GenericRecord>) record.get("attributes");
    var attrs = attrsRecords.stream().map(attr -> new Attr(attr.get(idPos).toString(),
      ((Integer) attr.get(quantityPos)).byteValue(),
      ((Integer) attr.get(amountPos)).byteValue(),
      (boolean) attr.get(activePos),
      (double) attr.get(percentPos),
      ((Integer) attr.get(sizePos)).shortValue())).toList();
    Utf8 name = (Utf8) record.get(namePos);
    Utf8 category = (Utf8) record.get(categoryPos);
    Utf8 country = (Utf8) record.get(countryPos);
    Type type = Type.valueOf(record.get(organizationTypePos).toString());
    organizations.add(new Org(name.toString(), category.toString(), country.toString(), type, attrs));
  }
}

The deserialization time drops to 7,353 ms, around 10% faster than the generated code version. Why? I don't have enough knowledge of its internals to venture an answer, but I was surprised by the result.

Avro Summary

	Generated Code	Generic Record	Optimized Generic Record
Serialization time	5,409 ms	5,903 ms	5,381 ms
Deserialization time	8,197 ms	8,471 ms	7,353 ms

Using GenericRecord allows us to gain some flexibility in the process without losing performance, but it makes the code much more verbose and prone to errors due to the manual mapping of fields..

The included dependencies are the same in all cases, and we can only save the generation of the source code.

Analysis and impressions

	JSON	Protocol Buffers	FlatBuffers	Avro
Serialization time	11,718 ms	5,823 ms	3,803 ms	5,409 ms
File size	2,457 MB	1,044 MB	600 MB	846 MB
GZ file size	525 MB	448 MB	414 MB	530 MB
Memory serializing	N/A	1.29 GB	0.6 GB - 1 GB	N/A
Deserialization time	20,410 ms	4,535 ms	202 - 1,876 ms	8,197 ms
Memory deserialization	2,193 MB	2,710 MB	0 - 600 MB	2,520 MB
JAR library size	1,910 KB	1,636 KB	64 KB	3,469 KB
Size of generated classes	N/A	40 KB	9 KB	36 KB

If we don't consider the optimizations applied in FlatBuffers, the Avro file takes up less space.
In the example, although all fields are required, you can easily make them nullable (consuming slightly more space).
For me, its main strength is that it doesn't require having all the data in memory to serialize the information. For example, you can read it from a database or a file, transform or enrich it according to your logic, and at the same time persist it in some type of OutputStream (created from a file or HTTP connection).
The possibility of defining a schema programmatically gives us the option of modifying the output format depending on the business logic and creating our own serialization tooling.
Avro is halfway between JSON and binary formats like Protocol Buffers and Flatbuffers:
- Avro supports flexible schema
- It is not necessary to know or agree on the schema beforehand to be able to read an Avro file
- In JSON you can not get the schema from the file itself (without fully parsing it first)
- Avro is more compact than JSON (but not human-readable)
- Serialization and deserialization time is faster than JSON
- You can easily serialize/deserialize all objects in a loop/stream without needing to have it all in memory

Java Serialization with Flatbuffers

Jerónimo López — Mon, 14 Nov 2022 18:37:26 +0000

In the previous post I analyzed Protocol Buffers format, using JSON as baseline. In this post I'm going to analyze FlatfBuffers and compare it with previously studied formats.

FlatBuffers was also created within Google in 2014 under Apache 2.0 license. It was developed to meet specific needs within the world of video games and mobile applications, where resources are more constrained.

Like Protocol Buffers, it relies on predefining the data format with similar syntax and generates a human-unreadable binary content. It has support for multiple languages and allows adding and deprecating fields without breaking compatibility.

The main distinction of FlatBuffers is that it implements zero-copy deserialization: it does not need to create objects or reserve new memory areas to parse the information, because it always works with the information in binary within a memory or disk area. The objects that represent the deserialized information do not contain the information, but know how to resolve the value when their get methods are called.

In the specific case of Java, this is not strictly correct, because for Strings you have to instantiate the char[] needed for its internal structure. But it is only necessary if you call the getter of the String type attribute. Only the information that is accessed is deserialized.

IDL and code generation

The file with the schema could be this one:

namespace com.jerolba.xbuffers.flat;

enum OrganizationType : byte { FOO, BAR, BAZ }

table Attribute {
    id: string;
    quantity: byte; 
    amount: byte;
    size: short;
    percent: double;
    active: bool;
}

table Organization {
    name: string;
    category: string;
    type: OrganizationType;
    country: string;
    attributes: [Attribute];
}

table Organizations {
    organizations: [Organization];
}

root_type Organizations;

To generate all Java classes you need to install the compiler flatc and execute it with some parameters referencing where the IDL file is located and the target path of the generated files:

 flatc -j -o /opt/src/main/java/  /opt/src/main/resources/organizations.fbs

As with Protocol Buffers, I prefer to use directly docker with an image ready to execute the command:

docker run --rm -v $(pwd)/src:/opt/src neomantra/flatbuffers flatc -j -o /opt/src/main/java/  /opt/src/main/resources/organizations.fbs

Serialization

This is the worst part of FlatBuffers: it's complex and it's not automatic. The serialization operation requires a manual coding process, where you describe step by step how to fill a binary buffer.

As you add elements of your data structures to the buffer, it returns offsets or pointers, which are the values used as references in the data structures that contain them. All in a recursive way.

If we see the data structure as a tree, we would have to do a postorder traversal.

The process is very fragile and it is very easy to make mistakes, so it will be necessary to cover this part with unit tests.

The code necessary to serialize the information starting from POJOs would look like this:

var organizations = dataFactory.getOrganizations(400_000)
FlatBufferBuilder builder = new FlatBufferBuilder();

int[] orgsArr = new int[organizations.size()];
int contOrgs = 0;
for (Org org : organizations) {
  int[] attributes = new int[org.attributes().size()];
  int contAttr = 0;
  for (Attr attr : org.attributes()) {
    int idOffset = builder.createString(attr.id());
    attributes[contAttr++] = Attribute.createAttribute(builder, 
        idOffset, attr.quantity(), attr.amount(), attr.size(), 
        attr.percent(), attr.active());
  }
  int attrsOffset = Organization.createAttributesVector(builder, attributes);

  int nameOffset = builder.createString(org.name());
  int categoryOffset = builder.createString(org.category());
  byte type = (byte) org.type().ordinal();
  int countryOffset = builder.createString(org.country());
  orgsArr[contOrgs++] = Organization.createOrganization(builder, 
    nameOffset, categoryOffset, type, countryOffset, attrsOffset)
}
int organizationsOffset = Organizations.createOrganizationsVector(builder, orgsArr);
int root_table = Organizations.createOrganizations(builder, organizationsOffset);
builder.finish(root_table);

try (var os = new FileOutputStream("/tmp/flatbuffer.json")) {
  InputStream sizedInputStream = builder.sizedInputStream();
  sizedInputStream.transferTo(os);
}

As you can see, it's a pretty ugly code, and you can easily make a mistake.

Serialization time: 5,639 ms
File size: 1,284 MB
Compressed file size: 530 MB
Memory required: internally creates a ByteBuffer that grows by power of two. To fit the 1.2 GB of data you need to allocate 2 GB of memory, unless you initially set it to 1.2 GB if you know its size beforehand.
Library size (flatbuffers-java-1.12.0.jar): 64,873 bytes
Size of generated classes: 9,080 bytes

Deserialization

Deserialization is simpler than serialization, and the only difficulty lies in instantiating a ByteBuffer containing the binary serialized information.

Depending on the requirements we can bring all the content into memory or use a memory-mapped file.

Once deserialized (or rather read the file), the objects you use do not actually contain the information, they are simply a proxy that knows how to locate it on demand. Therefore, if some data is not accessed, it is not really deserialized... on the other hand, every time you access the same value, you are deserializing it.

Reading the entire file into memory

If the available memory allows it and you are going to use the data intensively, you are more likely to want to read everything in memory:

try (RandomAccessFile file = new RandomAccessFile("/tmp/organizations.flatbuffers", "r")) {
  FileChannel inChannel = file.getChannel();
  ByteBuffer buffer = ByteBuffer.allocate((int) inChannel.size());
  inChannel.read(buffer);
  inChannel.close();
  buffer.flip();
  Organizations organizations = Organizations.getRootAsOrganizations(buffer);
  Organization organization = organizations.organizations(0);
  String name = organization.name();
  for (int i=0; i < organization.attributesLength(); i++){
    String attrId = organization.attributes(i).id();
  ......

Deserialization time accessing some attributes: 640 ms
Deserialization time accessing all attributes: 2,184 ms
Memory required: loading the entire file into memory, 1,284 MB

Mapping the file into memory

The Java FileChannel system abstracts whether all the information is directly in memory or whether it is read from disk as needed:

try (RandomAccessFile file = new RandomAccessFile("/tmp/organizations.flatbuffers", "r")) {
  FileChannel inChannel = file.getChannel();
  MappedByteBuffer buffer = inChannel.map(MapMode.READ_ONLY, 0, inChannel.size());
  buffer.load();
  Organizations organizations = Organizations.getRootAsOrganizations(buffer);
    .....
  inChannel.close();
}

Deserialization time accessing some attributes: 306 ms
Deserialization time accessing all attributes: 2,044 ms
Memory required: as it only creates file reading objects and temporary buffers, I do not know how to measure it, and I think it can be considered negligible.

I am surprised that using a memory-mapped file takes slightly less time. Probably as I run the benchmark multiple times, the operating system has cached the file in memory. If I were rigorous with the test I should change the process, but it is good enough to get an idea.

Analysis and impressions

	JSON	Protocol Buffers	FlatBuffers
Serialization time	11,718 ms	5,823 ms	5,639 ms
File size	2,457 MB	1,044 MB	1,284 MB
GZ file size	525 MB	448 MB	530 MB
Memory serializing	N/A	1.29 GB	1.3 GB - 2 GB
Deserialization time	20,410 ms	4,535 ms	306 - 2,184 ms
Memory deserialization	2,193 MB	2,710 MB	0 - 1,284 MB
JAR library size	1,910 KB	1,636 KB	64 KB
Size of generated classes	N/A	40 KB	9 KB

From the data and from what I have been able to see by playing with the formats, we can conclude:

FlatBuffers is highly recommended when deserialization time and memory consumption is important. This explains why it is so widely used in the world of video games and mobile applications. Its serialization API is very delicate and prone to errors.
In FlatBuffers, in an example like mine where there is a lot of data, in the serialization process it is important to configure a buffer size close to the final result, otherwise, the library will have to continuously extend the buffer, spending more memory and time.
In FlatBuffers, the JAR dependency needed to serialize and deserialize, together with the generated classes, is ridiculously small.
While in JSON and Protocol Buffers you need to deserialize all the information to access a part of it, in FlatBuffers you can randomly access any element without having to traverse and parse all the information that precedes it.
Protocol Buffers takes less space than FlatBuffers because the overhead per data structure (message/table) is smaller, and especially because Protocol Buffers serializes scalars with the value that takes the least bytes, compressing the information.
For scalar values, FlatBuffers does not support null. If a value is not present it takes the default value of the corresponding primitive (0, 0.0, or false). But Strings can be null.
If you need to represent the null value for a scalar, you can define a struct: struct NullableInt32 { i:int32 }. This adds more complexity to the serialization code and consumes more bytes.
If you follow the serialization algorithm, all normalized strings (countries and categories) from the original object graph will appear repeated in the serialized file.

Iterating serialization implementation

Implementing the serialization with FlatBuffers is really hard, but on the other hand, if you understand how it works internally, this problem can become a great advantage that works in its favor.

Every time you do something like this:

int idOffset = builder.createString(attr.id());

you are adding to the buffer a character array, and getting a kind of pointer or offset to use in the object that contains it.

If the same String is repeated in a serialization, can I reuse the pointer each time it appears? Yes.

If in a serialization you know that the same String will be repeated multiple times you can reuse its offset without breaking the internal representation of the serialization.

With a small change every time we serialize a String, using a Map<String, Integer> and querying it every time we add a String, we would have something like this:

var organizations = dataFactory.getOrganizations(400_000)
FlatBufferBuilder builder = new FlatBufferBuilder();
Map<String, Integer> strOffsets = new HashMap<>();   //<-- offsets cache

int[] orgsArr = new int[organizations.size()];
int contOrgs = 0;
for (Org org : organizations) {
  int[] attributes = new int[org.attributes().size()];
  int contAttr = 0;
  for (Attr attr : org.attributes()) {
    int idOffset = strOffsets.computeIfAbsent(attr.id(), builder::createString);  // <--
    attributes[contAttr++] = Attribute.createAttribute(builder, 
        idOffset, attr.quantity(), attr.amount(), attr.size(),
        attr.percent(), attr.active());
  }
  int attrsOffset = Organization.createAttributesVector(builder, attributes);

  int nameOffset = strOffsets.computeIfAbsent(org.name(), builder::createString);  // <--
  int categoryOffset = strOffsets.computeIfAbsent(org.category(), builder::createString);  // <--
  byte type = (byte) org.type().ordinal();
  int countryOffset = strOffsets.computeIfAbsent(org.country(), builder::createString);  // <--
  orgsArr[contOrgs++] = Organization.createOrganization(builder, nameOffset,
      categoryOffset, type, countryOffset, attrsOffset);
}
int organizationsOffset = Organizations.createOrganizationsVector(builder, orgsArr);
int root_table = Organizations.createOrganizations(builder, organizationsOffset);
builder.finish(root_table);

try (var os = new FileOutputStream("/tmp/flatbuffer.json")) {
  InputStream sizedInputStream = builder.sizedInputStream();
  sizedInputStream.transferTo(os);
}

Serialization time: 3,803 ms
File size: 600 MB
Compressed file size: 414 MB
Memory needed: to store those 600 MB it will use a ByteBuffer that will require 1 GB if you do not preconfigure it.

In this synthetic example, the reduction of the file size reaches more than 50%, and despite having to constantly look up a Map, the time it takes to serialize is significantly lower.

Because the format does not change, the deserialization retains the same properties and the read code does not change. If you choose to read the entire file into memory, you will gain the same memory space, while if you read it as a mapped file, the change will not be noticeable except for the lower I/O usage.

If we add this optimization to the comparison table, we have:

	JSON	Protocol Buffers	FlatBuffers	FlatBuffers V2
Serialization time	11,718 ms	5,823 ms	5,639 ms	3,803 ms
File size	2,457 MB	1,044 MB	1,284 MB	600 MB
GZ file size	525 MB	448 MB	530 MB	414 MB
Memory serializing	N/A	1.29 GB	1.3 GB - 2 GB	0.6 GB - 1 GB
Deserialization time	20,410 ms	4,535 ms	306 - 2,184 ms	202 - 1,876 ms
Memory deserialization	2,193 MB	2,710 MB	0 - 1,284 MB	0 - 600 MB

What we are actually doing is compressing the information at serialization time, taking advantage of the fact that we have some knowledge about the data. We are creating a compression dictionary manually.

Iterating again the idea

Can this process of String compression or normalization be applied to data structures that you serialize?

What if the tuple of attribute values were repeated between Organizations, can we reuse it? Yes

We can reuse the pointer of a previously serialized tuple to be referenced from the serialization of another organization. We would only need to create a Map where the key is the Attr class and the value is the offset of its first occurrence: Map<Attr, Integer> attrOffsets = new HashMap<>().

The example I have created for this article is synthetic and the values used are random. But in my real case, with data structures similar to the example, data is more repetitive and data reuse is higher.

In our real use case, we load the information from a MongoDB collection, and the numbers are these:

The records in MongoDB occupy 737 MB (which compresses on disk with WiredTiger, taking physically 510 MB)
If we serialize the same records with FlatBuffers without optimizations, it consumes 390 MB
If we further optimize by normalizing Strings, it takes 186 MB
If we normalize the Attr instances, the file requires 45 MB
If we compress the file with Gzip, the resulting file is only 25 MB

Conclusion

Knowing how the technologies you use work, their strengths and weaknesses, and the principles they are based on, allows you to optimize their use and push them beyond the usual use cases you are told about.

Knowing how your data looks like is also important when your data volume is high and your resources are limited. How can you structure your data to make it lighter? How can you structure your data to make it faster to process?

In our case, these optimizations have allowed us to postpone a complex architecture change. In an online service, instead of taking 60 seconds to load a batch of information, we can reduce to 3-4 seconds.

If the volume of data becomes really Big Data, I would advise you to use formats where these optimizations are already built-in, such as dictionaries or grouping data into columns.

Java Serialization with Protocol Buffers

Jerónimo López — Tue, 01 Nov 2022 10:42:28 +0000

At Clarity AI, we generate batches of data that our application has to load and process to show our clients the social impact information of many companies. By volume of information, it is not Big Data, but it is enough to be a problem to read and load it efficiently in processes with online users.

The information can be stored in a database or as files, serialized in a standard format and with a schema agreed with your Data Engineering team. Depending on your information and requirements, it can be as simple as CSV, XML or JSON, or Big Data formats such as Parquet, Avro, ORC, Arrow, or message serialization formats like Protocol Buffers, FlatBuffers, MessagePack, Thrift, or Cap'n Proto.

My idea is to analyze some of these systems over a collection of articles, using JSON as a reference, to compare all with something everyone knows.

To allow anyone to reach their own conclusions according to their requirements, I will analyze different technical aspects: creation time, size of the resulting file, size of the file on compression (gzip), the memory needed to create the file, size of the libraries used, deserialization time or the memory needed to parse and access the data.

In my case, because my access pattern is Write Once Read Many, in my final selection read factors will take preference over write factors.

Data model

Given this data model of organizations and their attributes, using new Java records:

record Org(String name, String category, String country, Type type,
           List<Attr> attributes) { }

record Attr(String id, byte quantity, byte amount, boolean active, 
            double percent, short size) { }

enum Type { FOO, BAR, BAZ }

I will simulate a heavy scenario, where each Organization will randomly have between 40 and 70 different Attributes, and in total, we will have about 400K organizations. The values of the attributes are also randomized.

JSON

It is the data interchange format par excellence and the facto standard in web services communication. Although it was defined as a format in the early 2000s, it was not until 2013 that the Ecma published the first version, which became an international standard in 2017.

With a simple syntax, it does not need to predefine the schema to be able to parse it. Because it is a plain text-based format, it is human readable and there are a lot of libraries to process it in all languages.

Serialization

Using a library like Jackson and without special annotations, the code is very simple:

var organizations = dataFactory.getOrganizations(400_000);

ObjectMapper mapper = new ObjectMapper();
try (var os = new FileOutputStream("/tmp/organizations.json")) {
  mapper.writeValue(organizations, os);
}

Metrics:

Serialization time: 11,718 ms
File size: 2,457 MB
Compressed file size: 525 MB
Memory required: because we are serializing directly to OutputStream, it consumes nothing apart from the required internal IO buffers.
Library size (jackson-xxx.jar): 1,956,679 bytes

Deserialization

It is difficult to define how to measure deserialization in this analysis. The objective is to recover the state of the entities from their binary representation, but which entities? the original class or the one provided by the tool with a similar interface?

To exploit the capabilities and strengths of each tool, we will keep the representation of the entity generated by each library. In the case of JSON will be the original class, but in other libraries will be a different class.

Jackson makes it very easy again and solves the problem with only 3 lines of code:

try (InputStream is = new FileInputStream("/tmp/organizations.json")) {
  ObjectMapper mapper = new ObjectMapper();
  List<Org> organizations mapper.readValue(is, new TypeReference<List<Org>>() {});
  ....
}

Deserialization time: 20,410 ms
Memory required: because is reconstructing the original object structures, they consume 2,193 MB

Protocol Buffers

Protocol Buffers is the system that Google developed internally to serialize data in a way that is both CPU and storage efficient, especially if you compare it with how it was done in those days with XML. In 2008 it was released under BSD license.

It is based on predefining what format the data will have through an IDL (Interface Definition Language) and from it, generating the source code that will be able to either write and read files with data. The producer and the consumer must somehow share the format defined in the IDL.

The format is flexible enough to support adding new fields and deprecating existing fields without breaking compatibility.

The information generated after serialization is an array of bytes, unreadable to humans.

Support for different programming languages is provided by the existence of a code generator for each language. If a language is not officially supported you can always find an implementation from the community.

Google has standardized it and made it the base of their server-to-server communication mechanism: gRPC, instead of the usual REST with JSON.

Although the documentation discourages its use with large datasets and proposes to split large collections into a "concatenation" of individual objects serialization, I'm going to evaluate and try it.

IDL and code generation

The file with the schema could be this:

syntax = "proto3";

package com.jerolba.xbuffers.protocol;

option java_multiple_files = true;
option java_package = "com.jerolba.xbuffers.protocol";
option java_outer_classname = "OrganizationsCollection";

message Organization {
  string name = 1;
  string category = 2;
  OrganizationType type = 3;
  string country = 4;
  repeated Attribute attributes = 5;

  enum OrganizationType {
    FOO = 0;
    BAR = 1;
    BAZ = 2;
  }

  message Attribute {
    string id = 1;
    int32 quantity = 2;
    int32 amount = 3;
    int32 size = 4;
    double percent = 5;
    bool active = 6;
  }

}

message Organizations {
  repeated Organization organizations = 1;
}

To generate all Java classes you need to install the compiler protoc and execute it with some parameters referencing where the IDL file is located and the target path of the generated files:

protoc --java_out=./src/main/java -I=./src/main/resources ./src/main/resources/organizations.proto

You have all the instructions here and you can download the compiler utility from here, but I prefer to use directly docker with an image ready to execute the command:

 docker run --rm -v $(pwd):$(pwd) -w $(pwd) znly/protoc --java_out=./src/main/java -I=./src/main/resources ./src/main/resources/organizations.proto

Serialization

Protocol Buffers does not directly serialize your POJOs, but you need to copy the information to the objects generated by the schema compiler.

The code needed to serialize the information from the POJOs would look like this:

var organizations = dataFactory.getOrganizations(400_000)

var orgsBuilder = Organizations.newBuilder();
for (Org org : organizations) {
    var organizationBuilder = Organization.newBuilder()
        .setName(org.name())
        .setCategory(org.category())
        .setCountry(org.country())
        .setType(OrganizationType.forNumber(org.type().ordinal()));
    for (Attr attr : org.attributes()) {
        var attribute = Attribute.newBuilder()
            .setId(attr.id())
            .setQuantity(attr.quantity())
            .setAmount(attr.amount())
            .setActive(attr.active())
            .setPercent(attr.percent())
            .setSize(attr.size())
            .build();
        organizationBuilder.addAttributes(attribute);
    }
    orgsBuilder.addOrganizations(organizationBuilder.build());
}
Organizations orgsBuffer = orgsBuilder.build();
try (var os = new FileOutputStream("/tmp/organizations.protobuffer")) {
  orgsBuffer.writeTo(os);
}

The code is verbose, but simple. If for some reason you decided to make your business logic work directly with the Protocol Buffers generated classes, all that copy code would be unnecessary.

Serialization time: 5,823 ms
File size: 1,044 MB
Compressed file size: 448 MB
Memory required: instantiating all these intermediate objects in memory before serializing them requires 1,315 MB
Library size (protobuf-java-3.16.0.jar): 1,675,739 bytes
Size of generated classes: 41,229 bytes

Deserialization

It is also quite straightforward, and it is enough to pass an InputStream to rebuild in memory the whole object graph:

try (InputStream is = new FileInputStream("/tmp/organizations.protobuffer")) {
    Organizations organizations = Organizations.parseFrom(is);
    .....
}

The objects are instances of the classes generated from the schema, not the original records.

Deserialization time: 4,535 ms
Memory required: reconstructing the object structures defined by the schema takes up 2,710 MB

Analysis and impressions

	JSON	Protocol Buffers
Serialization time	11,718 ms	5,823 ms
File size	2,457 MB	1,044 MB
GZ file size	525 MB	448 MB
Memory serializing	N/A	1.29 GB
Deserialization time	20,410 ms	4,535 ms
Memory deserialization	2,193 MB	2,710 MB
JAR library size	1,910 KB	1,636 KB
Size of generated classes	N/A	40 KB

From the data and from what I have been able to see by playing with the formats, we can conclude:

JSON is undoubtedly slower and heavier, but it is a very comfortable and convenient system.
Protocol Buffers is a good choice: fast serialization and deserialization, and with compact and compressible files. Its API is simple and intuitive, and the format is widely used, becoming a market standard with multiple use cases.
Both formats are all or nothing: you need to deserialize all the information to access a part of it. While in other tools you can access any element without having to traverse and parse all the information that precedes it.
Protocol Buffers uses 32-bit integers for all integer types (int, short, byte), but when serializing represents them with the value that takes the least bytes, saving some bytes.
For the same reason, when Protocol Buffers generates the classes from the IDL, it defines everything as 32-bit integers and deserializes the values to int32. That is why the memory consumption of deserialized objects is 23% higher than JSON. What it gains by compressing integers it loses in memory consumption.
For scalar values, Protocol Buffers does not support null. If a value is not present it takes the default value of the corresponding primitive (0, 0.0, or false), while Null Strings are deserialized to "". This article proposes different strategies to simulate null values.
Although in the original object graph the name of the countries or categories are Strings that are only present in memory once, when serialized to a binary format they will appear as many times as references you have, and when deserialized you will have as many instances as references you had. That is why in both cases, the memory consumed when deserializing occupies twice the memory occupied by the objects originally serialized¹.

¹ Since Java 8u20, thanks to JEP 192, we have an option in G1 to de-duplicate Strings during garbage collection. But it is disabled by default and when it is enabled we have no control over when it is executed, so we can't rely on that optimization to reduce the size of deserialization.