Export Deephaven tables to Parquet files

The Deephaven Parquet Python module provides tools to integrate Deephaven with the Parquet file format. This document covers writing Deephaven tables to single Parquet files, key-value partitioned Parquet directories, and flat partitioned Parquet directories. You can write each layout to local storage, S3, or Google Cloud Storage.

By default, Deephaven writes Parquet files with SNAPPY compression. To choose a different codec or set other write options, see Optional arguments.

First, create some tables to use in the examples in this guide.

Write to a single Parquet file

To local storage

Write a Deephaven table to a single Parquet file with parquet.write. Pass the table as the table argument and the destination file path as the path argument. The path must end with the .parquet file extension.

To use a different compression codec, set the compression_codec_name argument. Optional arguments lists the available codecs.

Write _metadata and _common_metadata files by setting the generate_metadata_files argument to True. These files hold the schema and other metadata for the Parquet files that the write produces, and Deephaven places them in the destination directory. Readers use them to find the data files and the full schema without listing the directory tree, so they matter most for the partitioned Parquet directories described later in this guide. See Metadata-partitioned directories for what each file contains.

To S3

Use parquet.write to write Deephaven tables to Parquet files on S3. The path should be the URI of the destination file in S3. Supply an instance of the S3Instructions class to the special_instructions argument to specify the details of the connection to the S3 instance. See Special instructions (S3 and GCS) for the S3Instructions arguments that apply when writing. The example defines s3_instructions once. The later S3 examples in this guide reuse it.

Partitioned Parquet directories

Deephaven can also write tables to a directory of Parquet files instead of a single file. It supports two directory layouts:

• A key-value partitioned directory nests its Parquet files in subdirectories named key=value, one level per partitioning column. A partitioning column's values name the subdirectories, such as Class=Math, instead of being stored inside each file.
• A flat partitioned directory holds its Parquet files side by side in a single directory. It has no partitioning columns.

When a query filters on a partitioning column, Deephaven can skip the subdirectories the filter excludes.

Write to a key-value partitioned Parquet directory

A key-value partitioned directory stores each partition in a key=value subdirectory, as described in Partitioned Parquet directories. Writing the grades table with Class as the partitioning column produces the subdirectories Class=Math, Class=Science, and Class=History.

You can write a key-value partitioned directory from a regular Deephaven table or from a partitioned table. When you write a partitioned table, its key columns become the partitioning columns.

Define a table definition

A regular table needs a table definition, which is the table's schema with the partitioning columns marked. A regular table's own definition usually has no partitioning columns, so the write fails unless you pass a definition that has them. Pass it as the table_definition argument to write_partitioned.

One way to build a table definition is as a list of ColumnDefinition objects. Create each one with col_def, passing the column's name and a type from the deephaven.dtypes module. To mark a column as a partitioning column, set the column_type argument of col_def to ColumnType.PARTITIONING.

Create a table definition for the grades table defined above.

To local storage

Pass a regular table or a partitioned table to the table argument of parquet.write_partitioned, and set the destination_dir argument to the root directory for the partitioned Parquet data. Deephaven creates any missing directories in the path.

Set the generate_metadata_files argument to True to write _metadata and _common_metadata files at the root of the directory, as described in Write to a single Parquet file.

To S3

Use parquet.write_partitioned to write key-value partitioned Parquet directories to S3. The destination_dir should be the URI of the destination directory in S3. Use the s3_instructions defined in Write to a single Parquet file as the special_instructions argument.

Write to a flat partitioned Parquet directory

A flat partitioned directory holds one Parquet file per table, as described in Partitioned Parquet directories.

To local storage

Use parquet.write or parquet.batch_write to write Deephaven tables to Parquet files in flat partitioned directories.

Call parquet.write once per table, giving each file its own path in the same directory, as in Write to a single Parquet file.

Use parquet.batch_write to accomplish the same thing by passing multiple tables to the tables argument and multiple destination paths to the paths argument. If the tables have different definitions, also pass the table_definition argument.

To write a Deephaven partitioned table to a flat partitioned Parquet directory, look up each constituent table by key with get_constituent, so that each file name matches its class, and pass the tables to parquet.batch_write.

To S3

Use parquet.batch_write to write a list of Deephaven tables to a flat partitioned Parquet directory in S3. The paths should be the URIs of the destination files in S3. Use the s3_instructions defined in Write to a single Parquet file as the special_instructions argument.

Optional arguments

The write, write_partitioned, and batch_write functions from the Deephaven Parquet Python module all accept the following optional arguments, which control how Deephaven writes data to Parquet:

• table_definition: The table definition or schema, provided as a TableDefinition, a dictionary of string-DType pairs, or a list of ColumnDefinition instances. When not provided, Deephaven uses the column definitions of the table or tables being written.
• col_instructions: A list of ColumnInstruction objects that customize how particular columns are written. See Column instructions. The default is None.
• compression_codec_name: The name of the compression codec to use for the whole file. Codecs trade write speed against compression ratio. The options are:
  • SNAPPY: A codec based on Google's Snappy compression format that aims for high speed and reasonable compression. This is the default.
  • UNCOMPRESSED: No compression.
  • LZ4_RAW: A codec based on the LZ4 block format.
  • LZO: A codec based on or interoperable with the LZO compression library.
  • GZIP: A codec based on the GZIP format defined by RFC 1952. This differs from the related zlib and deflate formats.
  • ZSTD: A codec with a high compression ratio, based on the Zstandard format defined by RFC 8478.
  • BROTLI: A codec based on Brotli, offering high compression ratios. Deephaven doesn't include a Brotli codec. To use this option, add a Brotli codec (org.apache.hadoop.io.compress.BrotliCodec) to the server classpath. Without one, the write fails with a Failed to find CompressionCodec error.
  • LZ4: Deprecated. Use LZ4_RAW instead.
• max_dictionary_keys: The maximum number of unique keys the writer should add to a dictionary page before switching to non-dictionary encoding. Dictionary-based encoding stores each column's unique values in dictionary pages. The writer applies this limit only to String columns and columns of String arrays or vectors, and ignores it for a column whose use_dictionary hint is True. Defaults to 2^20 (1,048,576).
• max_dictionary_size: The maximum number of bytes the writer should add to the dictionary before switching to non-dictionary encoding. The writer applies this limit only to String columns and columns of String arrays or vectors, and ignores it for a column whose use_dictionary hint is True. Defaults to 2^20 (1,048,576).
• target_page_size: The target page size in bytes. Defaults to 65,536 bytes (64 KiB), which you can change with the Parquet.defaultTargetPageSize configuration property.
• generate_metadata_files: Whether to generate Parquet _metadata and _common_metadata files. Defaults to False.
• row_group_info: Sets how the written data is divided into row groups, the horizontal slices of rows that make up a Parquet file. The available RowGroupInfo options are:
  • RowGroupInfo.single_group: All data is within a single row group. This is the default.
  • RowGroupInfo.max_rows(max_rows): Splits the data into row groups of no more than max_rows rows each.
  • RowGroupInfo.max_groups(num_row_groups): Splits the data into the requested number of row groups of nearly equal size. If the row count is not evenly divisible by that number, some row groups contain one fewer row.
  • RowGroupInfo.by_groups(groups, max_rows): Writes the rows for each unique combination of values in the groups columns as their own row group. groups is a list of column names. The rows for each group must be contiguous in the table, or the write raises an error. The optional max_rows argument splits any group with more rows than max_rows into smaller row groups, as RowGroupInfo.max_rows does.
• index_columns: The sets of columns to write data indexes for, as a sequence of sequences. For example, [["Col1"], ["Col1", "Col2"]] writes an index on Col1 and an index on Col1 and Col2. Deephaven writes each index to its own Parquet file under a .dh_metadata/indexes subdirectory of the data file's directory. If the source table lacks a listed index, Deephaven computes it. The default depends on the function:
  • write: Writes the indexes on the source table.
  • write_partitioned with a regular table: Writes the indexes on the source table, except single-column indexes on a partitioning column.
  • batch_write: Writes the indexes on the first table.
  • write_partitioned with a partitioned table: Writes no indexes.
• special_instructions: Connection details for writing to S3 or Google Cloud Storage, provided as an S3Instructions instance. See Special instructions (S3 and GCS).

write_partitioned also accepts a base_name argument, which sets the names of the Parquet files it writes in each partition directory.

Column instructions

The col_instructions argument to write, write_partitioned, and batch_write must be a list of ColumnInstruction instances. Each ColumnInstruction maps a column in the Deephaven table to a column in the resulting Parquet files, and optionally sets an object codec or dictionary encoding for that column.

ColumnInstruction has the following arguments that apply when writing:

• column_name: The name of the Deephaven column that these instructions apply to. Required when writing.
• parquet_column_name: The name of the corresponding column in the Parquet dataset.
• codec_name: The fully qualified name of an ObjectCodec class that serializes the column's values to and from bytes. Use it for types that have no language-agnostic Parquet representation. This is not the compression codec. The compression_codec_name argument sets that for the whole file.
• codec_args: An implementation-specific argument string passed to the codec named by codec_name.
• use_dictionary: A hint to keep dictionary-based encoding for this column. The writer already tries dictionary encoding for every String column and every column of String arrays or vectors. Setting the hint to True removes the max_dictionary_keys and max_dictionary_size limits for the column, so the writer never falls back to non-dictionary encoding. The writer holds the column's dictionary and encoded pages in memory until it finishes the column, so setting the hint on a high-cardinality String column can exhaust the JVM heap. Set it only when the number of unique values is bounded. The writer ignores the hint for other column types. The default is False.

Special instructions (S3 and GCS)

The special_instructions argument to write, write_partitioned, and batch_write is relevant when writing to S3, an S3-compatible store, or Google Cloud Storage (gs:// URIs), and takes an instance of the S3Instructions class. This class specifies details for connecting to the store.

The following S3Instructions arguments are relevant when writing. The API documentation lists the rest, including arguments that apply only to reads.

• region_name: The region name of the AWS S3 bucket. If you don't set it, the AWS SDK looks for a region in these places, in order:

  • The aws.region system property
  • The AWS_REGION environment variable
  • The {user.home}/.aws/credentials and {user.home}/.aws/config files
  • The EC2 metadata service, when running on EC2

  If none of these gives the bucket's correct region, Deephaven looks up the region with one additional request.

• credentials: The credentials object for authenticating to the S3 instance. The default is Credentials.resolving.

• endpoint_override: The endpoint to connect to. Connections to AWS rarely need it. Set it when connecting to a non-AWS, S3-compatible API. The default is None.

• connection_timeout: Time to wait for a successful S3 connection before timing out. The default is 2 seconds.

• write_timeout: The amount of time to wait when writing a fragment before timing out. The default is 2 seconds.

• write_part_size: The part or chunk size, in bytes, when writing to S3. The default is 10 MiB, and the minimum is 5,242,880 bytes (5 MiB).

• num_concurrent_write_parts: The maximum number of parts that can be uploaded concurrently without blocking. The default is 64. This value cannot exceed max_concurrent_requests. Each write can hold up to this many parts in heap memory, each write_part_size bytes. The defaults allow about 640 MiB for one write, so raising either value increases heap use.

• max_concurrent_requests: The maximum number of concurrent requests to make to S3, for reads and writes. The default is 256.

• profile_name: The AWS profile name used to configure the default region, credentials, and other settings.

• config_file_path: The path to the AWS configuration file.

• credentials_file_path: The path to the AWS credentials file.

The access_key_id, secret_access_key, and anonymous_access arguments are deprecated. Use Credentials.basic(access_key_id, secret_access_key) or Credentials.anonymous for the credentials argument instead.

Related documentation

• Import Parquet files
• Parquet formats