Dataset discovery#

ADataset object can be created with thedataset() function. Wecan pass it the path to the directory containing the data files:

In [11]:importpyarrow.datasetasdsIn [12]:dataset=ds.dataset(base/"parquet_dataset",format="parquet")In [13]:datasetOut[13]:<pyarrow._dataset.FileSystemDataset at 0x7fe030eacd00>

In addition to searching a base directory,dataset() accepts a path to asingle file or a list of file paths.

Creating aDataset object does not begin reading the data itself. Ifneeded, it only crawls the directory to find all the files:

In [14]:dataset.filesOut[14]:['/tmp/pyarrow-9_qw6tb_/parquet_dataset/data1.parquet', '/tmp/pyarrow-9_qw6tb_/parquet_dataset/data2.parquet']

… and infers the dataset’s schema (by default from the first file):

In [15]:print(dataset.schema.to_string(show_field_metadata=False))a: int64b: doublec: int64

Using theDataset.to_table() method we can read the dataset (or a portionof it) into a pyarrow Table (note that depending on the size of your datasetthis can require a lot of memory, see below on filtering / iterative loading):

In [16]:dataset.to_table()Out[16]:pyarrow.Tablea: int64b: doublec: int64----a: [[0,1,2,3,4],[5,6,7,8,9]]b: [[1.1373880927195317,-0.5829723903558806,1.0058469280348412,-0.7464153609137238,-0.49764206660895927],[-0.9559528016600878,-2.007285802328415,-0.3980382178636434,1.8734612388530516,0.8336648980261276]]c: [[1,2,1,2,1],[2,1,2,1,2]]# converting to pandas to see the contents of the scanned tableIn [17]:dataset.to_table().to_pandas()Out[17]:   a         b  c0  0  1.137388  11  1 -0.582972  22  2  1.005847  13  3 -0.746415  24  4 -0.497642  15  5 -0.955953  26  6 -2.007286  17  7 -0.398038  28  8  1.873461  19  9  0.833665  2

Reading different file formats#

The above examples use Parquet files as dataset sources but the Dataset APIprovides a consistent interface across multiple file formats and filesystems.Currently, Parquet, ORC, Feather / Arrow IPC, and CSV file formats aresupported; more formats are planned in the future.

If we save the table as Feather files instead of Parquet files:

In [18]:importpyarrow.featherasfeatherIn [19]:feather.write_feather(table,base/"data.feather")

…then we can read the Feather file using the same functions, but with specifyingformat="feather":

In [20]:dataset=ds.dataset(base/"data.feather",format="feather")In [21]:dataset.to_table().to_pandas().head()Out[21]:   a         b  c0  0  1.137388  11  1 -0.582972  22  2  1.005847  13  3 -0.746415  24  4 -0.497642  1

Customizing file formats#

The format name as a string, like:

ds.dataset(...,format="parquet")

is short hand for a default constructedParquetFileFormat:

ds.dataset(...,format=ds.ParquetFileFormat())

TheFileFormat objects can be customized using keywords. For example:

parquet_format=ds.ParquetFileFormat(read_options={'dictionary_columns':['a']})ds.dataset(...,format=parquet_format)

Will configure column"a" to be dictionary encoded on scan.

Different partitioning schemes#

The above example uses a hive-like directory scheme, such as “/year=2009/month=11/day=15”.We specified this passing thepartitioning="hive" keyword. In this case,the types of the partition keys are inferred from the file paths.

It is also possible to explicitly define the schema of the partition keysusing thepartitioning() function. For example:

part=ds.partitioning(pa.schema([("year",pa.int16()),("month",pa.int8()),("day",pa.int32())]),flavor="hive")dataset=ds.dataset(...,partitioning=part)

“Directory partitioning” is also supported, where the segments in the file pathrepresent the values of the partition keys without including the name (thefield name are implicit in the segment’s index). For example, given field names“year”, “month”, and “day”, one path might be “/2019/11/15”.

Since the names are not included in the file paths, these must be specifiedwhen constructing a directory partitioning:

part=ds.partitioning(field_names=["year","month","day"])

Directory partitioning also supports providing a full schema rather than inferringtypes from file paths.

Customizing the batch size#

An iterative read of a dataset is often called a “scan” of the dataset and pyarrowuses an object called aScanner to do this. A Scanner is created for youautomatically by theto_table() andto_batches() method of the dataset.Any arguments you pass to these methods will be passed on to the Scanner constructor.

One of those parameters is thebatch_size. This controls the maximum size of thebatches returned by the scanner. Batches can still be smaller than thebatch_sizeif the dataset consists of small files or those files themselves consist of smallrow groups. For example, a parquet file with 10,000 rows per row group will yieldbatches with, at most, 10,000 rows unless thebatch_size is set to a smaller value.

The default batch size is one million rows and this is typically a good default butyou may want to customize it if you are reading a large number of columns.

Writing partitioned data#

A partitioning object can be used to specify how your output data should be partitioned.This uses the same kind of partitioning objects we used for reading datasets. To writeour above data out to a partitioned directory we only need to specify how we want thedataset to be partitioned. For example:

In [56]:part=ds.partitioning(   ....:pa.schema([("c",pa.int16())]),flavor="hive"   ....:)   ....:In [57]:ds.write_dataset(table,"partitioned_dataset",format="parquet",partitioning=part)

This will create two files. Half our data will be in the dataset_root/c=1 directory andthe other half will be in the dataset_root/c=2 directory.

Partitioning performance considerations#

Partitioning datasets has two aspects that affect performance: it increases the number offiles and it creates a directory structure around the files. Both of these have benefitsas well as costs. Depending on the configuration and the size of your dataset, the costscan outweigh the benefits.

Because partitions split up the dataset into multiple files, partitioned datasets can beread and written with parallelism. However, each additional file adds a little overhead inprocessing for filesystem interaction. It also increases the overall dataset size sinceeach file has some shared metadata. For example, each parquet file contains the schema andgroup-level statistics. The number of partitions is a floor for the number of files. Ifyou partition a dataset by date with a year of data, you will have at least 365 files. Ifyou further partition by another dimension with 1,000 unique values, you will have up to365,000 files. This fine of partitioning often leads to small files that mostly consist ofmetadata.

Partitioned datasets create nested folder structures, and those allow us to prune whichfiles are loaded in a scan. However, this adds overhead to discovering files in the dataset,as we’ll need to recursively “list directory” to find the data files. Too finepartitions can cause problems here: Partitioning a dataset by date for a years worthof data will require 365 list calls to find all the files; adding another column withcardinality 1,000 will make that 365,365 calls.

The most optimal partitioning layout will depend on your data, access patterns, and whichsystems will be reading the data. Most systems, including Arrow, should work across arange of file sizes and partitioning layouts, but there are extremes you should avoid. Theseguidelines can help avoid some known worst cases:

• Avoid files smaller than 20MB and larger than 2GB.

• Avoid partitioning layouts with more than 10,000 distinct partitions.

For file formats that have a notion of groups within a file, such as Parquet, similarguidelines apply. Row groups can provide parallelism when reading and allow data skippingbased on statistics, but very small groups can cause metadata to be a significant portionof file size. Arrow’s file writer provides sensible defaults for group sizing in most cases.

Configuring files open during a write#

When writing data to the disk, there are a few parameters that can beimportant to optimize the writes, such as the number of rows per file andthe maximum number of open files allowed during the write.

Set the maximum number of files opened with themax_open_files parameter ofwrite_dataset().

Ifmax_open_files is set greater than 0 then this will limit the maximumnumber of files that can be left open. This only applies to writing partitioneddatasets, where rows are dispatched to the appropriate file depending on theirpartition values. If an attempt is made to open too many files then the leastrecently used file will be closed. If this setting is set too low you may endup fragmenting your data into many small files.

If your process is concurrently using other file handlers, either with adataset scanner or otherwise, you may hit a system file handler limit. Forexample, if you are scanning a dataset with 300 files and writing out to900 files, the total of 1200 files may be over a system limit. (On Linux,this might be a “Too Many Open Files” error.) You can either reduce thismax_open_files setting or increase the file handler limit on yoursystem. The default value is 900 which allows some number of filesto be open by the scanner before hitting the default Linux limit of 1024.

Another important configuration used inwrite_dataset() ismax_rows_per_file.

Set the maximum number of rows written in each file with themax_rows_per_filesparameter ofwrite_dataset().

Ifmax_rows_per_file is set greater than 0 then this will limit how manyrows are placed in any single file. Otherwise there will be no limit and onefile will be created in each output directory unless files need to be closed to respectmax_open_files. This setting is the primary way to control file size.For workloads writing a lot of data, files can get very large without arow count cap, leading to out-of-memory errors in downstream readers. Therelationship between row count and file size depends on the dataset schemaand how well compressed (if at all) the data is.

Configuring rows per group during a write#

The volume of data written to the disk per each group can be configured.This configuration includes a lower and an upper bound.The minimum number of rows required to form a row group isdefined with themin_rows_per_group parameter ofwrite_dataset().

The maximum number of rows allowed per group is defined with themax_rows_per_group parameter ofwrite_dataset().

Ifmax_rows_per_group is set greater than 0 then the dataset writer may splitup large incoming batches into multiple row groups. If this value is set thenmin_rows_per_group should also be set or else you may end up with very smallrow groups (e.g. if the incoming row group size is just barely larger than this value).

Row groups are built into the Parquet and IPC/Feather formats but don’t affect JSON or CSV.When reading back Parquet and IPC formats in Arrow, the row group boundaries become therecord batch boundaries, determining the default batch size of downstream readers.Additionally, row groups in Parquet files have column statistics which can help readersskip irrelevant data but can add size to the file. As an extreme example, if one setsmax_rows_per_group=1 in Parquet, they will have large files because most of the fileswill be row group statistics.

Writing large amounts of data#

The above examples wrote data from a table. If you are writing a large amount of datayou may not be able to load everything into a single in-memory table. Fortunately, thewrite_dataset() method also accepts an iterable of record batches. This makes it reallysimple, for example, to repartition a large dataset without loading the entire datasetinto memory:

In [58]:old_part=ds.partitioning(   ....:pa.schema([("c",pa.int16())]),flavor="hive"   ....:)   ....:In [59]:new_part=ds.partitioning(   ....:pa.schema([("c",pa.int16())]),flavor=None   ....:)   ....:In [60]:input_dataset=ds.dataset("partitioned_dataset",partitioning=old_part)# A scanner can act as an iterator of record batches but you could also receive# data from the network (e.g. via flight), from your own scanning, or from any# other method that yields record batches.  In addition, you can pass a dataset# into write_dataset directly but this method is useful if you want to customize# the scanner (e.g. to filter the input dataset or set a maximum batch size)In [61]:scanner=input_dataset.scanner()In [62]:ds.write_dataset(scanner,"repartitioned_dataset",format="parquet",partitioning=new_part)

After the above example runs our data will be in dataset_root/1 and dataset_root/2directories. In this simple example we are not changing the structure of the data(only the directory naming schema) but you could also use this mechanism to changewhich columns are used to partition the dataset. This is useful when you expect toquery your data in specific ways and you can utilize partitioning to reduce theamount of data you need to read.

Customizing & inspecting written files#

By default the dataset API will create files named “part-i.format” where “i” is a integergenerated during the write and “format” is the file format specified in the write_datasetcall. For simple datasets it may be possible to know which files will be created but forlarger or partitioned datasets it is not so easy. Thefile_visitor keyword can be usedto supply a visitor that will be called as each file is created:

In [63]:deffile_visitor(written_file):   ....:print(f"path={written_file.path}")   ....:print(f"size={written_file.size} bytes")   ....:print(f"metadata={written_file.metadata}")   ....:

In [64]:ds.write_dataset(table,"dataset_visited",format="parquet",partitioning=part,   ....:file_visitor=file_visitor)   ....:path=dataset_visited/c=1/part-0.parquetsize=815 bytesmetadata=<pyarrow._parquet.FileMetaData object at 0x7fe02cd5c090>  created_by: parquet-cpp-arrow version 22.0.0  num_columns: 2  num_rows: 5  num_row_groups: 1  format_version: 2.6  serialized_size: 0path=dataset_visited/c=2/part-0.parquetsize=817 bytesmetadata=<pyarrow._parquet.FileMetaData object at 0x7fe02cd07830>  created_by: parquet-cpp-arrow version 22.0.0  num_columns: 2  num_rows: 5  num_row_groups: 1  format_version: 2.6  serialized_size: 0

This will allow you to collect the filenames that belong to the dataset and store them elsewherewhich can be useful when you want to avoid scanning directories the next time you need to readthe data. It can also be used to generate the _metadata index file used by other tools such asDask or Spark to create an index of the dataset.

Configuring format-specific parameters during a write#

In addition to the common options shared by all formats there are also format specific optionsthat are unique to a particular format. For example, to allow truncated timestamps while writingParquet files:

In [65]:parquet_format=ds.ParquetFileFormat()In [66]:write_options=parquet_format.make_write_options(allow_truncated_timestamps=True)In [67]:ds.write_dataset(table,"sample_dataset2",format="parquet",partitioning=part,   ....:file_options=write_options)   ....: