+
+
+
+
diff --git a/objects.inv b/objects.inv
index d8084e0123adc38ce47891f52a432752e210c178..b18fb9df311604730caecfa0cca1b0142c43f9b1 100644
GIT binary patch
delta 1588
zcmV-42Fv-F4ZaPKoqrigjr8EufL#PV5sbuLvynuBqSo2``cfaes}-r?%xK+9Vrcn3
zFGUSMHOf`tWFfz+&8P0KXZfvAH}&E7tdLD}^V^NA^@qIsMY!7N3+^Zs2P5p87Q
zlP+CZ_ha*FP1jv)0x)lbM3;ZdEO{T_Nvcir;!o_M%FD+c_g@S>>`>ML0-n&1>IT
z8|AKna>s$21%GPme9gc)$c3)_09>ld7ht3IGHctCFXwdfd##FmKIsi{5-mE6wEs3y
zpQLVlVnpe*(@**=B&lSnD8<{NI0g|?;^U9e3g@V$3J*1|b`O)rT3W~)8+1v#$xOA?
zyVgoK2Gqh(nFjT%l3X083@#^EO=fkywmLdrqmBd>Nq=8U>vVSu!p|XXIV>)Su&2`y
zp6;+iSw@eY3aQ~bYIU`jR?EsYsHh!P)J}DlQLFjYYQJQEw5C@6p$sd}u)DnWrRG{c
z*!!!MEH-!6%ER;6H3#|ah3Q!|T$v~m#I6rIsjEhGgVU)IvG|!hW#(7^-Q*=Mh8eWi
zg*s&w4}V9r*IAA-ip2%Z>qto1#bOujbQB{@R^8D=pR{6y0uTZpO<8JpFa1U@|HD_o
z&O43zk*R)&Z>ARKs}Vxnsb3WFEGANb{Wxv}R(M<@;$E}G+0sK8iU9~>pPOrvWJ*Xz
zWDq}C`@!5(afo>_kHfsaD9DNG)I9v~iMO`1!O
z}P#gj8u?$l>l<%J6L&G0Ku+`p{uU%l&UYeb_8l{pi8zevNH
zT2&rdw*&o}yvA82%j*zL>o~~ZBo=j;o;YB|9CE|wrYr*eY#%=4WK*ao_t42CpMODY
zNA~&J>;QriK=82y6~Jt=#j^y0mifdJ#{d)pv*OSeo{XVZVqw2cRr13Jxl1OK_|FsA
z%_7t*Hmcz!h4MU>w=P66_eo;*8#D~FOQ$yoLNO?C)Ou`z?NZpDB5U{g7`@KD1R0|NE#USdGoFd-M~G$<4B6(byOWD7ESvG|h^
z#7A>}u@)yWXc5T@#u|*$1U}3f#`6a)Xb?#`2qRRTkjmX~$THvpQ)4WPc`8+-4Z8
zZJ5Q(Fm>rwO(fqLYcxs`_{Dr^n$-mcw2kCWW35Ihj9)PWED!!+uI??62pY-pT}CdX
z@Y47*2;>&a^QD?i8Gu(T7no=@MF38dbYY@_&LM6j0SQeGGLmNO7FGXk;58cuQUFCkElg4e-50$maZ%H`h
z*B4cZQaV3f<^uJt-f9=rcQIFT@lULHE2ql1liRn0XiRG}^~P-&_k7zHoqFo3QP%x`
zPH*-6x?RlhmQ>UisruJ4`fJ<+pofCvrU3=@^JiK6jWeOfH2zOvY%Q+r_4v!5MchVS
mTmzp|OwoOD_WSVRYWmGOT*0V3VguC`c9<*QzxxlU-J`W5b`a_S
delta 1550
zcmV+p2J!j64VVp(oqq*MgY@9kfL#PV5sXCJY$Q>jC_9^9U+QCbwbE!fGg|kO7`uF*
zSE7c)56$ML2)fYU&F*9O*S-E$n5%mKdsgVCx%%x&*Y;iBeWOD6)w^GQy?Xm52Shtv
z1mioY4#BgOSz&b5)~Yl?=Q`*P^3YaUu&&~PywSmk%zaL4W>W3!_?Zn@xlbYVS(b
z^`FY(KZSb~lO9b_I?t8vrSdC!vQ2P)FV+`M$BSsxv++%*-Nd;FwDY->{7jxQiwpQ}@=|AMhU|5wPFca>
zNcK9(Nqm|(K$jwbq~l_(UD5cuHA(uPku?$z=?;wAX`pfNu(
z)2H}mYH|7$Aiz%jqDW>jks|EdaU-z7;~W8d%@${C4`FDAAeduru1S(PAqmJZ{$lNi
zbC1n2=Gi=s^ZKSJYrb+!@D9x}CMT1eD-u-xbAL+a3t)AcAg(8f8(p*J(nEWcCYx`T
zVuFbvGlP2E>bu#qrs;(a#|x&hxio38p`(%kd2A@?B0~2(>!LMEv^K*K6jR9^O_Z?3
zt+#PXwaVZ6b
zx-@+ex%(LZ|8ROKIX$5on^5Ju_h0X3M}Ki#!rx#1y#I&BKsW8P8=3a`Dh1lobLA_$
zb_NCupcLj|yo$TvynZyFb(I%pp>Kw7Y2yBUbN}|-Ru4`!rZ8D>(ECLm*43tp$hsZq
zm*h3U8(m(8=-MDbE>2=ehv|s}R>~naer_rv@Xz+~LryV;dvcGROv)L|eiWat&3_Ic
zI01woOK<_gE?YfIU}%L;JaG)bAqX!V+rqOk+>2P*Z*!IM@FDJ!*(CY%M0T?T^@fe=
za+AV&9_w2drdawUDfRaYBXo04L7^dt=8*gS%&8r)OJ}@OLZEQ3HXK#N6TzrQ7+ZuC`K)SvR$gd
zC{5s(WxaI%phXRUx?d{4SuUFvc(^W@>TpsgO_0)Ij@t}FwOy7O3rs_L*?$2PBvXw>
zDFMG)u*|bM!=Sc6aWmCwl;ZdeDSACSFNThis is the documentation for the native Rust/Python implementation of Delta Lake. It is based on the delta-rs Rust library and requires no Spark or JVM dependencies. For the PySpark implementation, see delta-spark instead.
This module provides the capability to read, write, and manage Delta Lake tables with Python or Rust without Spark or Java. It uses Apache Arrow under the hood, so is compatible with other Arrow-native or integrated libraries such as pandas, DuckDB, and Polars.
\"Rust deltalake\" refers to the Rust API of delta-rs (no Spark dependency)
\"Python deltalake\" refers to the Python API of delta-rs (no Spark dependency)
\"Delta Spark\" refers to the Scala impementation of the Delta Lake transaction log protocol. This depends on Spark and Java.
"},{"location":"#why-implement-the-delta-lake-transaction-log-protocol-in-rust-and-scala","title":"Why implement the Delta Lake transaction log protocol in Rust and Scala?","text":"
Delta Spark depends on Java and Spark, which is fine for many use cases, but not all Delta Lake users want to depend on these libraries. delta-rs allows using Delta Lake in Rust or other native projects when using a JVM is often not an option.
Python deltalake lets you query Delta tables without depending on Java/Scala.
Suppose you want to query a Delta table with pandas on your local machine. Python deltalake makes it easy to query the table with a simple pip install command - no need to install Java.
Refers to the Databricks Unity Catalog <https://docs.databricks.com/data-governance/unity-catalog/index.html>_
","boost":2},{"location":"api/delta_writer/","title":"Writer","text":"","boost":10},{"location":"api/delta_writer/#write-to-delta-tables","title":"Write to Delta Tables","text":"","boost":10},{"location":"api/delta_writer/#deltalake.write_deltalake","title":"deltalake.write_deltalake","text":"
If the table does not already exist, it will be created.
This function only supports writer protocol version 2 currently. When attempting to write to an existing table with a higher min_writer_version, this function will throw DeltaProtocolError.
Note that this function does NOT register this table in a data catalog.
A locking mechanism is needed to prevent unsafe concurrent writes to a delta lake directory when writing to S3. DynamoDB is the only available locking provider at the moment in delta-rs. To enable DynamoDB as the locking provider, you need to set the AWS_S3_LOCKING_PROVIDER to 'dynamodb' as a storage_option or as an environment variable.
Additionally, you must create a DynamoDB table with the name 'delta_rs_lock_table' so that it can be automatically discovered by delta-rs. Alternatively, you can use a table name of your choice, but you must set the DYNAMO_LOCK_TABLE_NAME variable to match your chosen table name. The required schema for the DynamoDB table is as follows:
Data to write. If passing iterable, the schema must also be given.
required schemaOptional[Union[Schema, Schema]]
Optional schema to write.
Nonepartition_byOptional[Union[List[str], str]]
List of columns to partition the table by. Only required when creating a new table.
NonefilesystemOptional[FileSystem]
Optional filesystem to pass to PyArrow. If not provided will be inferred from uri. The file system has to be rooted in the table root. Use the pyarrow.fs.SubTreeFileSystem, to adopt the root of pyarrow file systems.
How to handle existing data. Default is to error if table already exists. If 'append', will add new data. If 'overwrite', will replace table with new data. If 'ignore', will not write anything if table already exists.
Optional write options for Parquet (ParquetFileWriteOptions). Can be provided with defaults using ParquetFileWriteOptions().make_write_options(). Please refer to https://github.com/apache/arrow/blob/master/python/pyarrow/_dataset_parquet.pyx#L492-L533 for the list of available options. Only used in pyarrow engine.
Nonemax_partitionsOptional[int]
the maximum number of partitions that will be used. Only used in pyarrow engine.
Nonemax_open_filesint
Limits the maximum number of files that can be left open while writing. If an attempt is made to open too many files then the least recently used file will be closed. If this setting is set too low you may end up fragmenting your data into many small files. Only used in pyarrow engine.
1024max_rows_per_fileint
Maximum number of rows per file. If greater than 0 then this will limit how many rows are placed in any single file. Otherwise there will be no limit and one file will be created in each output directory unless files need to be closed to respect max_open_files min_rows_per_group: Minimum number of rows per group. When the value is set, the dataset writer will batch incoming data and only write the row groups to the disk when sufficient rows have accumulated. Only used in pyarrow engine.
10 * 1024 * 1024max_rows_per_groupint
Maximum number of rows per group. If the value is set, then the dataset writer may split up large incoming batches into multiple row groups. If this value is set, then min_rows_per_group should also be set.
A Writer Properties instance for the Rust parquet writer.
Create a Writer Properties instance for the Rust parquet writer, see options https://arrow.apache.org/rust/parquet/file/properties/struct.WriterProperties.html:
Parameters:
Name Type Description Default data_page_size_limitOptional[int]
Limit DataPage size to this in bytes.
Nonedictionary_page_size_limitOptional[int]
Limit the size of each DataPage to store dicts to this amount in bytes.
Nonedata_page_row_count_limitOptional[int]
Limit the number of rows in each DataPage.
Nonewrite_batch_sizeOptional[int]
Splits internally to smaller batch size.
Nonemax_row_group_sizeOptional[int]
Max number of rows in row group.
NonecompressionOptional[str]
compression type
Nonecompression_levelOptional[int]
level of compression, only relevant for subset of compression types
None","boost":10},{"location":"api/delta_writer/#convert-to-delta-tables","title":"Convert to Delta Tables","text":"","boost":10},{"location":"api/delta_writer/#deltalake.convert_to_deltalake","title":"deltalake.convert_to_deltalake","text":"
Currently only HIVE partitioned tables are supported. Convert to delta creates a transaction log commit with add actions, and additional properties provided such as configuration, name, and description.
Parameters:
Name Type Description Default uriUnion[str, Path]
URI of a table.
required partition_byOptional[Schema]
Optional partitioning schema if table is partitioned.
Nonepartition_strategyOptional[Literal['hive']]
Optional partition strategy to read and convert
NonemodeLiteral['error', 'ignore']
How to handle existing data. Default is to error if table already exists. If 'ignore', will not convert anything if table already exists.
","boost":2},{"location":"api/schema/","title":"Schema","text":"","boost":2},{"location":"api/schema/#schema-and-field","title":"Schema and field","text":"
Schemas, fields, and data types are provided in the deltalake.schema submodule.
get schema with all variable size types (list, binary, string) as large variants (with int64 indices). This is for compatibility with systems like Polars that only support the large versions of Arrow types.
Create the Delta Table from a path with an optional version. Multiple StorageBackends are currently supported: AWS S3, Azure Data Lake Storage Gen2, Google Cloud Storage (GCS) and local URI. Depending on the storage backend used, you could provide options values using the storage_options parameter.
Parameters:
Name Type Description Default table_uriUnion[str, Path, PathLike[str]]
the path of the DeltaTable
required versionOptional[int]
version of the DeltaTable
Nonestorage_optionsOptional[Dict[str, str]]
a dictionary of the options to use for the storage backend
Nonewithout_filesbool
If True, will load table without tracking files. Some append-only applications might have no need of tracking any files. So, the DeltaTable will be loaded with a significant memory reduction.
Falselog_buffer_sizeOptional[int]
Number of files to buffer when reading the commit log. A positive integer. Setting a value greater than 1 results in concurrent calls to the storage api. This can decrease latency if there are many files in the log since the last checkpoint, but will also increase memory usage. Possible rate limits of the storage backend should also be considered for optimal performance. Defaults to 4 * number of cpus.
Delete expired log files before current version from table. The table log retention is based on the configuration.logRetentionDuration value, 30 days by default.
How to handle existing data. Default is to error if table already exists. If 'append', returns not support error if table exists. If 'overwrite', will CREATE_OR_REPLACE table. If 'ignore', will not do anything if table already exists. Defaults to \"error\".
Delete records from a Delta Table that statisfy a predicate.
When a predicate is not provided then all records are deleted from the Delta Table. Otherwise a scan of the Delta table is performed to mark any files that contain records that satisfy the predicate. Once files are determined they are rewritten without the records.
Parameters:
Name Type Description Default predicateOptional[str]
a SQL where clause. If not passed, will delete all rows.
Get the list of files as absolute URIs, including the scheme (e.g. \"s3://\").
Local files will be just plain absolute paths, without a scheme. (That is, no 'file://' prefix.)
Use the partition_filters parameter to retrieve a subset of files that match the given filters.
Parameters:
Name Type Description Default partition_filtersOptional[List[Tuple[str, str, Any]]]
the partition filters that will be used for getting the matched files
None
Returns:
Type Description List[str]
list of the .parquet files with an absolute URI referenced for the current version of the DeltaTable
Predicates are expressed in disjunctive normal form (DNF), like [(\"x\", \"=\", \"a\"), ...]. DNF allows arbitrary boolean logical combinations of single partition predicates. The innermost tuples each describe a single partition predicate. The list of inner predicates is interpreted as a conjunction (AND), forming a more selective and multiple partition predicates. Each tuple has format: (key, op, value) and compares the key with the value. The supported op are: =, !=, in, and not in. If the op is in or not in, the value must be a collection such as a list, a set or a tuple. The supported type for value is str. Use empty string '' for Null partition value.
The paths are as they are saved in the delta log, which may either be relative to the table root or absolute URIs.
Parameters:
Name Type Description Default partition_filtersOptional[List[Tuple[str, str, Any]]]
the partition filters that will be used for getting the matched files
None
Returns:
Type Description List[str]
list of the .parquet files referenced for the current version of the DeltaTable
Predicates are expressed in disjunctive normal form (DNF), like [(\"x\", \"=\", \"a\"), ...]. DNF allows arbitrary boolean logical combinations of single partition predicates. The innermost tuples each describe a single partition predicate. The list of inner predicates is interpreted as a conjunction (AND), forming a more selective and multiple partition predicates. Each tuple has format: (key, op, value) and compares the key with the value. The supported op are: =, !=, in, and not in. If the op is in or not in, the value must be a collection such as a list, a set or a tuple. The supported type for value is str. Use empty string '' for Null partition value.
Name Type Description Default data_catalogDataCatalog
the Catalog to use for getting the storage location of the Delta Table
required database_namestr
the database name inside the Data Catalog
required table_namestr
the table name inside the Data Catalog
required data_catalog_idOptional[str]
the identifier of the Data Catalog
NoneversionOptional[int]
version of the DeltaTable
Nonelog_buffer_sizeOptional[int]
Number of files to buffer when reading the commit log. A positive integer. Setting a value greater than 1 results in concurrent calls to the storage api. This can decrease latency if there are many files in the log since the last checkpoint, but will also increase memory usage. Possible rate limits of the storage backend should also be considered for optimal performance. Defaults to 4 * number of cpus.
Add actions represent the files that currently make up the table. This data is a low-level representation parsed from the transaction log.
Parameters:
Name Type Description Default flattenbool
whether to flatten the schema. Partition values columns are given the prefix partition., statistics (null_count, min, and max) are given the prefix null_count., min., and max., and tags the prefix tags.. Nested field names are concatenated with ..
False
Returns:
Type Description RecordBatch
a PyArrow RecordBatch containing the add action data.
Load/time travel a DeltaTable to a specified version number, or a timestamp version of the table. If a string is passed then the argument should be an RFC 3339 and ISO 8601 date and time string format.
Parameters:
Name Type Description Default versionUnion[int, str, datetime]
the identifier of the version of the DeltaTable to load
Time travel Delta table to the latest version that's created at or before provided datetime_string argument. The datetime_string argument should be an RFC 3339 and ISO 8601 date and time string.
Deprecated
Load_version and load_with_datetime have been combined into DeltaTable.load_as_version.
Parameters:
Name Type Description Default datetime_stringstr
the identifier of the datetime point of the DeltaTable to load
Pass the source data which you want to merge on the target delta table, providing a predicate in SQL query like format. You can also specify on what to do when the underlying data types do not match the underlying table.
Parameters:
Name Type Description Default sourceUnion[Table, RecordBatch, RecordBatchReader, Dataset, DataFrame]
source data
required predicatestr
SQL like predicate on how to merge
required source_aliasOptional[str]
Alias for the source table
Nonetarget_aliasOptional[str]
Alias for the target table
Noneerror_on_type_mismatchbool
specify if merge will return error if data types are mismatching :default = True
Repair the Delta Table by auditing active files that do not exist in the underlying filesystem and removes them. This can be useful when there are accidental deletions or corrupted files.
Active files are ones that have an add action in the log, but no corresponding remove action. This operation creates a new FSCK transaction containing a remove action for each of the missing or corrupted files.
Parameters:
Name Type Description Default dry_runbool
when activated, list only the files, otherwise add remove actions to transaction log. Defaults to False.
False
Returns: The metrics from repair (FSCK) action.
Example
from deltalake import DeltaTable\ndt = DeltaTable('TEST')\ndt.repair(dry_run=False)\n
Build a pandas dataframe using data from the DeltaTable.
Parameters:
Name Type Description Default partitionsOptional[List[Tuple[str, str, Any]]]
A list of partition filters, see help(DeltaTable.files_by_partitions) for filter syntax
NonecolumnsOptional[List[str]]
The columns to project. This can be a list of column names to include (order and duplicates will be preserved)
NonefilesystemOptional[Union[str, FileSystem]]
A concrete implementation of the Pyarrow FileSystem or a fsspec-compatible interface. If None, the first file path will be used to determine the right FileSystem
NonefiltersOptional[FilterType]
A disjunctive normal form (DNF) predicate for filtering rows. If you pass a filter you do not need to pass partitions
Build a PyArrow Dataset using data from the DeltaTable.
Parameters:
Name Type Description Default partitionsOptional[List[Tuple[str, str, Any]]]
A list of partition filters, see help(DeltaTable.files_by_partitions) for filter syntax
NonefilesystemOptional[Union[str, FileSystem]]
A concrete implementation of the Pyarrow FileSystem or a fsspec-compatible interface. If None, the first file path will be used to determine the right FileSystem
Build a PyArrow Table using data from the DeltaTable.
Parameters:
Name Type Description Default partitionsOptional[List[Tuple[str, str, Any]]]
A list of partition filters, see help(DeltaTable.files_by_partitions) for filter syntax
NonecolumnsOptional[List[str]]
The columns to project. This can be a list of column names to include (order and duplicates will be preserved)
NonefilesystemOptional[Union[str, FileSystem]]
A concrete implementation of the Pyarrow FileSystem or a fsspec-compatible interface. If None, the first file path will be used to determine the right FileSystem
NonefiltersOptional[FilterType]
A disjunctive normal form (DNF) predicate for filtering rows. If you pass a filter you do not need to pass partitions
Update a matched table row based on the rules defined by updates. If a predicate is specified, then it must evaluate to true for the row to be updated.
Parameters:
Name Type Description Default updatesDict[str, str]
a mapping of column name to update SQL expression.
Updating all source fields to target fields, source and target are required to have the same field names. If a predicate is specified, then it must evaluate to true for the row to be updated.
Parameters:
Name Type Description Default predicateOptional[str]
Update a target row that has no matches in the source based on the rules defined by updates. If a predicate is specified, then it must evaluate to true for the row to be updated.
Parameters:
Name Type Description Default updatesDict[str, str]
a mapping of column name to update SQL expression.
Insert a new row to the target table based on the rules defined by updates. If a predicate is specified, then it must evaluate to true for the new row to be inserted.
Parameters:
Name Type Description Default updatesdict
a mapping of column name to insert SQL expression.
Insert a new row to the target table, updating all source fields to target fields. Source and target are required to have the same field names. If a predicate is specified, then it must evaluate to true for the new row to be inserted.
Parameters:
Name Type Description Default predicateOptional[str]
Compacts small files to reduce the total number of files in the table.
This operation is idempotent; if run twice on the same table (assuming it has not been updated) it will do nothing the second time.
If this operation happens concurrently with any operations other than append, it will fail.
Parameters:
Name Type Description Default partition_filtersOptional[FilterType]
the partition filters that will be used for getting the matched files
Nonetarget_sizeOptional[int]
desired file size after bin-packing files, in bytes. If not provided, will attempt to read the table configuration value delta.targetFileSize. If that value isn't set, will use default value of 256MB.
Nonemax_concurrent_tasksOptional[int]
the maximum number of concurrent tasks to use for file compaction. Defaults to number of CPUs. More concurrent tasks can make compaction faster, but will also use more memory.
minimum interval in seconds or as timedeltas before a new commit is created. Interval is useful for long running executions. Set to 0 or timedelta(0), if you want a commit per partition.
Nonewriter_propertiesOptional[WriterProperties]
Pass writer properties to the Rust parquet writer.
None
Returns:
Type Description Dict[str, Any]
the metrics from optimize
Example
Use a timedelta object to specify the seconds, minutes or hours of the interval.
Reorders the data using a Z-order curve to improve data skipping.
This also performs compaction, so the same parameters as compact() apply.
Parameters:
Name Type Description Default columnsIterable[str]
the columns to use for Z-ordering. There must be at least one column. partition_filters: the partition filters that will be used for getting the matched files
required target_sizeOptional[int]
desired file size after bin-packing files, in bytes. If not provided, will attempt to read the table configuration value delta.targetFileSize. If that value isn't set, will use default value of 256MB.
Nonemax_concurrent_tasksOptional[int]
the maximum number of concurrent tasks to use for file compaction. Defaults to number of CPUs. More concurrent tasks can make compaction faster, but will also use more memory.
Nonemax_spill_sizeint
the maximum number of bytes to spill to disk. Defaults to 20GB.
minimum interval in seconds or as timedeltas before a new commit is created. Interval is useful for long running executions. Set to 0 or timedelta(0), if you want a commit per partition.
Nonewriter_propertiesOptional[WriterProperties]
Pass writer properties to the Rust parquet writer.
None
Returns:
Type Description Dict[str, Any]
the metrics from optimize
Example
Use a timedelta object to specify the seconds, minutes or hours of the interval.
The Parquet file stores the data that was written. The _delta_log directory stores metadata about the transactions. Let's inspect the _delta_log/00000000000000000000.json file.
This transaction adds a data file and marks the two exising data files for removal. Marking a file for removal in the transaction log is known as \"tombstoning the file\" or a \"logical delete\". This is different from a \"physical delete\" which actually removes the data file from storage.
"},{"location":"how-delta-lake-works/architecture-of-delta-table/#how-delta-table-operations-differ-from-data-lakes","title":"How Delta table operations differ from data lakes","text":"
Data lakes consist of data files persisted in storage. They don't have a transaction log that retain metadata about the transactions.
Data lakes perform transactions differently than Delta tables.
When you perform an overwrite tranasction with a Delta table, you logically delete the exiting data without physically removing it.
Data lakes don't support logical deletes, so you have to physically delete the data from storage.
Logical data operations are safer because they can be rolled back if they don't complete successfully. Physically removing data from storage can be dangerous, especially if it's before a transaction is complete.
We're now ready to look into Delta Lake ACID transactions in more detail.
"},{"location":"integrations/delta-lake-arrow/","title":"Delta Lake Arrow Integrations","text":"
Delta Lake tables can be exposed as Arrow tables and Arrow datasets, which allows for interoperability with a variety of query engines.
This page shows you how to convert Delta tables to Arrow data structures and teaches you the difference between Arrow tables and Arrow datasets. Tables are \"eager\" and datasets are \"lazy\", which has important performance implications, keep reading to learn more!
"},{"location":"integrations/delta-lake-arrow/#delta-lake-to-arrow-dataset","title":"Delta Lake to Arrow Dataset","text":"
Delta tables can easily be exposed as Arrow datasets. This makes it easy for any query engine that can read Arrow datasets to read a Delta table.
Let's take a look at the h2o groupby dataset that contains 9 columns of data. Here are three representative rows of data:
Arrow datasets allow for the predicates to get pushed down to the query engine, so the query is executed quickly.
"},{"location":"integrations/delta-lake-arrow/#delta-lake-to-arrow-table","title":"Delta Lake to Arrow Table","text":"
You can also run the same query with DuckDB on an Arrow table:
quack = duckdb.arrow(table.to_pyarrow_table())\nquack.filter(\"id1 = 'id016' and v2 > 10\")\n
This returns the same result, but it runs slower.
"},{"location":"integrations/delta-lake-arrow/#difference-between-arrow-dataset-and-arrow-table","title":"Difference between Arrow Dataset and Arrow Table","text":"
Arrow Datasets are lazy and allow for full predicate pushdown unlike Arrow tables which are eagerly loaded into memory.
The previous DuckDB queries were run on a 1 billion row dataset that's roughly 50 GB when stored as an uncompressed CSV file. Here are the runtimes when the data is stored in a Delta table and the queries are executed on a 2021 Macbook M1 with 64 GB of RAM:
Arrow table: 17.1 seconds
Arrow dataset: 0.01 seconds
The query runs much faster on an Arrow dataset because the predicates can be pushed down to the query engine and lots of data can be skipped.
Arrow tables are eagerly materialized in memory and don't allow for the same amount of data skipping.
"},{"location":"integrations/delta-lake-arrow/#multiple-query-engines-can-query-arrow-datasets","title":"Multiple query engines can query Arrow Datasets","text":"
Other query engines like DataFusion can also query Arrow datasets, see the following example:
from datafusion import SessionContext\n\nctx = SessionContext()\nctx.register_dataset(\"my_dataset\", table.to_pyarrow_dataset())\nctx.sql(\"select * from my_dataset where v2 > 5\")\n
Delta tables can easily be exposed as Arrow tables/datasets.
Therefore any query engine that can read an Arrow table/dataset can also read a Delta table.
Arrow datasets allow for more predicates to be pushed down to the query engine, so they can perform better performance than Arrow tables.
"},{"location":"integrations/delta-lake-datafusion/","title":"Using Delta Lake with DataFusion","text":"
This page explains how to use Delta Lake with DataFusion.
Delta Lake offers DataFusion users better performance and more features compared to other formats like CSV or Parquet.
Delta Lake works well with the DataFusion Rust API and the DataFusion Python API. It's a great option for all DataFusion users.
Delta Lake also depends on DataFusion to implement SQL-related functionality under the hood. We will also discuss this dependency at the end of this guide in case you're interested in learning more about the symbiotic relationship between the two libraries.
"},{"location":"integrations/delta-lake-datafusion/#delta-lake-performance-benefits-for-datafusion-users","title":"Delta Lake performance benefits for DataFusion users","text":"
Let's run some DataFusion queries on a Parquet file and a Delta table with the same data to learn more about the performance benefits of Delta Lake.
Suppose you have the following dataset with 1 billion rows and 9 columns. Here are the first three rows of data:
ctx.sql(\"select id1, sum(v1) as v1 from my_delta_table where id1='id096' group by id1\")\n
That query takes 2.8 seconds to execute.
Let's register the same dataset as a Parquet table, run the same query, and compare the runtime difference.
Register the Parquet table and run the query:
path = \"G1_1e9_1e2_0_0.parquet\"\nctx.register_parquet(\"my_parquet_table\", path)\nctx.sql(\"select id1, sum(v1) as v1 from my_parquet_table where id1='id096' group by id1\")\n
This query takes 5.3 seconds to run.
Parquet stores data in row groups and DataFusion can intelligently skip row groups that don't contain relevant data, so the query is faster than a file format like CSV which doesn't support row group skipping.
Delta Lake stores file-level metadata information in the transaction log, so it can skip entire files when queries are executed. Delta Lake can skip entire files and then skip row groups within the individual files. This makes Delta Lake even faster than Parquet files, especially for larger datasets spread across many files.
"},{"location":"integrations/delta-lake-datafusion/#delta-lake-features-for-datafusion-users","title":"Delta Lake features for DataFusion users","text":"
Delta Lake also provides other features that are useful for DataFusion users like ACID transactions, concurrency protection, time travel, versioned data, and more.
"},{"location":"integrations/delta-lake-datafusion/#why-delta-lake-depends-on-datafusion","title":"Why Delta Lake depends on DataFusion","text":"
Delta Lake depends on DataFusion to provide some end-user features.
DataFusion is useful in providing SQL-related Delta Lake features. Some examples:
Update and merge are written in terms of SQL expressions.
Invariants and constraints are written in terms of SQL expressions.
Anytime we have to evaluate SQL, we need some sort of SQL engine. We use DataFusion for that.
Delta Lake is a great file format for DataFusion users.
Delta Lake also uses DataFusion to provide some end-user features.
DataFusion and Delta Lake have a wonderful symbiotic relationship and play very nicely with each other.
See this guide for more information on Delta Lake and PyArrow and why PyArrow Datasets are often a better option than PyArrow tables.
"},{"location":"integrations/delta-lake-pandas/","title":"Using Delta Lake with pandas","text":"
Delta Lake is a great storage system for pandas analyses. This page shows how it's easy to use Delta Lake with pandas, the unique features Delta Lake offers pandas users, and how Delta Lake can make your pandas analyses run faster.
Delta Lake is very easy to install for pandas analyses, just run pip install deltalake.
Delta Lake allows for performance optimizations, so pandas queries can run much faster than the query run on data stored in CSV or Parquet. See the following chart for the query runtime for the a Delta tables compared with CSV/Parquet.
Z Ordered Delta tables run this query much faster than when the data is stored in Parquet or CSV. Let's dive in deeper and see how Delta Lake makes pandas faster.
"},{"location":"integrations/delta-lake-pandas/#delta-lake-makes-pandas-queries-run-faster","title":"Delta Lake makes pandas queries run faster","text":"
There are a few reasons Delta Lake can make pandas queries run faster:
column pruning: only grabbing the columns relevant for a query
file skipping: only reading files with data for the query
row group skipping: only reading row groups with data for the query
Z ordering data: colocating similar data in the same files, so file skipping is more effective
Reading less data (fewer columns and/or fewer rows) is how Delta Lake makes pandas queries run faster.
Parquet allows for column pruning and row group skipping, but doesn't support file-level skipping or Z Ordering. CSV doesn't support any of these performance optimizations.
Let's take a look at a sample dataset and run a query to see the performance enhancements offered by Delta Lake.
Suppose you have a 1 billion row dataset with 9 columns, here are the first three rows of the dataset:
Parquet stores data in row groups and allows for skipping when the filters predicates are set. Run the Parquet query again with row group skipping enabled:
Here are the contents after the overwrite operation (version 2 of the Delta table):
+-------+----------+\n| num | letter |\n|-------+----------|\n| 8 | dd |\n| 9 | ee |\n+-------+----------+\n
Read in the Delta table and it will grab the latest version by default:
DeltaTable(\"tmp/some-table\").to_pandas()\n\n+-------+----------+\n| num | letter |\n|-------+----------|\n| 11 | aa |\n| 22 | bb |\n+-------+----------+\n
You can easily time travel back to version 0 of the Delta table:
DeltaTable(\"tmp/some-table\", version=0).to_pandas()\n\n+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n| 3 | c |\n+-------+----------+\n
You can also time travel to version 1 of the Delta table:
DeltaTable(\"tmp/some-table\", version=1).to_pandas()\n\n+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n| 3 | c |\n| 8 | dd |\n| 9 | ee |\n+-------+----------+\n
Time travel is a powerful feature that pandas users cannot access with CSV or Parquet.
Delta tables only allow you to append DataFrame with matching schema by default. Suppose you have a DataFrame with num and animal columns, which is different from the Delta table that has columns with num and letter columns.
Try to append this DataFrame with a mismatched schema to the existing table:
This transaction will be rejected and will return the following error message:
ValueError: Schema of data does not match table schema\nData schema:\nnum: int64\nanimal: string\n-- schema metadata --\npandas: '{\"index_columns\": [{\"kind\": \"range\", \"name\": null, \"start\": 0, \"' + 474\nTable Schema:\nnum: int64\nletter: string\n
Schema enforcement protects your table from getting corrupted by appending data with mismatched schema. Parquet and CSV don't offer schema enforcement for pandas users.
"},{"location":"integrations/delta-lake-pandas/#overwriting-schema-of-table","title":"Overwriting schema of table","text":"
You can overwrite the table contents and schema by setting the overwrite_schema option. Here's how to overwrite the table contents:
Here are the contents of the table after the values and schema have been overwritten:
+-------+----------+\n| num | animal |\n|-------+----------|\n| 5 | cat |\n| 6 | dog |\n+-------+----------+\n
"},{"location":"integrations/delta-lake-pandas/#in-memory-vs-in-storage-data-changes","title":"In-memory vs. in-storage data changes","text":"
It's important to distinguish between data stored in-memory and data stored on disk when understanding the functionality offered by Delta Lake.
pandas loads data from storage (CSV, Parquet, or Delta Lake) into in-memory DataFrames.
pandas makes it easy to modify the data in memory, say update a column value. It's not easy to update a column value in storage systems like CSV or Parquet using pandas.
Delta Lake makes it easy for pandas users to update data in storage.
"},{"location":"integrations/delta-lake-pandas/#why-delta-lake-allows-for-faster-queries","title":"Why Delta Lake allows for faster queries","text":"
Delta tables store data in many files and metadata about the files in the transaction log. Delta Lake allows for certain queries to skip entire files, which makes pandas queries run much faster.
Delta Lake provides many features that make it an excellent format for pandas analyses:
performance optimizations make pandas queries run faster
data management features make pandas analyses more reliable
advanced features allow you to perform more complex pandas analyses
Python deltalake offers pandas users a better experience compared with CSV/Parquet.
"},{"location":"integrations/delta-lake-polars/","title":"Using Delta Lake with polars","text":"
This page explains why Delta Lake is a great storage system for Polars analyses.
You will learn how to create Delta tables with Polars, how to query Delta tables with Polars, and the unique advantages Delta Lake offers the Polars community.
Here are some amazing benefits that Delta Lake provides Polars users:
time travel
ACID transactions for reliable writes
better performance with file skipping
enhanced file skipping via Z Ordering
ability to rollback mistakes
and many, many more
Let's start by showing how to use Polars with Delta Lake, explore how Delta Lake can make Polars queries run faster, and then look at all the cool features Delta Lake offers Polars users.
"},{"location":"integrations/delta-lake-polars/#creating-a-delta-lake-table-with-polars","title":"Creating a Delta Lake table with Polars","text":"
Create a Polars DataFrame and write it out to a Delta table:
import polars as pl\n\ndf = pl.DataFrame({\"x\": [1, 2, 3]})\ndf.write_delta(\"tmp/bear_delta_lake\")\n
This dataset is 50GB when stored in an uncompressed CSV file. Let's run some queries on this dataset when it's stored in different file formats with Polars.
This section will show the runtime for a query when the data is stored in CSV, Parquet, and Delta Lake and explain why Delta tables are the fastest.
Start by running a query on an uncompressed CSV file with read_csv:
This query runs in 8.3 seconds. It's much faster because Polars is optimized to skip row groups in Parquet files that don't contain data that's relevant for the query.
This query runs in 7.2 seconds. Polars can run this query faster because it can inspect the Delta transaction log and skip entire files that don't contain relevant data before performing the ordinary Parquet row group skipping.
Finally run the query on the Delta table after it has been Z Ordered by id1:
This query runs in 3.5 seconds. The query on the Z Ordered Delta table is even faster because similar data has been co-located in the same files. This allows for even greater data skipping.
Polars can leverage file skipping to query Delta tables very quickly.
"},{"location":"integrations/delta-lake-polars/#why-polars-is-fast-with-delta-lake","title":"Why Polars is fast with Delta Lake","text":"
Delta tables consist of metadata in a transaction log and data stored in Parquet files.
When Polars queries a Delta table, it starts by consulting the transaction log to understand the metadata of each file in the Delta table. This allows for Polars to quickly identify which files should be skipped by the query.
CSV files don't contain any such metadata, so file skipping isn't an option. Polars can skip Parquet files based on metadata, but it needs to open up each file and read the metadata, which is slower that grabbing the file-level metadata directly from the transaction log.
Parquet doesn't allow users to easily Z Order the data and colocate similar data in the same row groups. The Z Order optimizations are only supported in Delta tables.
Delta Lake offers Polars users with unique performance optimizations.
"},{"location":"integrations/delta-lake-polars/#other-delta-lake-features-relevant-for-polars-users","title":"Other Delta Lake features relevant for Polars users","text":"
This guide shows how Delta Lake is a great storage format for Polars analyses.
Delta Lake is easy to use, fast, and full of features that are great for Polars users.
"},{"location":"usage/","title":"Usage","text":"
A DeltaTable represents the state of a delta table at a particular version. This includes which files are currently part of the table, the schema of the table, and other metadata such as creation time.
Python Rust
DeltaTable
from deltalake import DeltaTable\n\ndt = DeltaTable(\"../rust/tests/data/delta-0.2.0\")\nprint(f\"Version: {dt.version()}\")\nprint(f\"Files: {dt.files()}\")\n
DeltaTable
let table = deltalake::open_table(\"../rust/tests/data/simple_table\").await.unwrap();\nprintln!(\"Version: {}\", table.version());\nprintln!(\"Files: {}\", table.get_files());\n
"},{"location":"usage/appending-overwriting-delta-lake-table/","title":"Appending to and overwriting a Delta Lake table","text":"
This section explains how to append to an exising Delta table and how to overwrite a Delta table.
"},{"location":"usage/appending-overwriting-delta-lake-table/#delta-lake-append-transactions","title":"Delta Lake append transactions","text":"
Suppose you have a Delta table with the following contents:
+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n| 3 | c |\n+-------+----------+\n
Here are the contents of the Delta table after the overwrite operation:
+-------+----------+\n| num | letter |\n|-------+----------|\n| 11 | aa |\n| 22 | bb |\n+-------+----------+\n
Overwriting just performs a logical delete. It doesn't physically remove the previous data from storage. Time travel back to the previous version to confirm that the old version of the table is still accessable.
dt = DeltaTable(\"tmp/some-table\", version=1)\n\n+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n| 3 | c |\n| 8 | dd |\n| 9 | ee |\n+-------+----------+\n
"},{"location":"usage/create-delta-lake-table/","title":"Creating a Delta Lake Table","text":"
This section explains how to create a Delta Lake table.
You can easily write a DataFrame to a Delta table.
from deltalake import write_deltalake\nimport pandas as pd\n\ndf = pd.DataFrame({\"num\": [1, 2, 3], \"letter\": [\"a\", \"b\", \"c\"]})\nwrite_deltalake(\"tmp/some-table\", df)\n
Here are the contents of the Delta table in storage:
+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n| 3 | c |\n+-------+----------+\n
"},{"location":"usage/deleting-rows-from-delta-lake-table/","title":"Deleting rows from a Delta Lake table","text":"
This section explains how to delete rows from a Delta Lake table.
Suppose you have the following Delta table with four rows:
+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n| 3 | c |\n| 4 | d |\n+-------+----------+\n
Here's how to delete all the rows where the num is greater than 2:
Here are the contents of the Delta table after the delete operation has been performed:
+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n+-------+----------+\n
"},{"location":"usage/examining-table/","title":"Examining a Table","text":""},{"location":"usage/examining-table/#metadata","title":"Metadata","text":"
The delta log maintains basic metadata about a table, including:
A unique id
A name, if provided
A description, if provided
The list of partitionColumns.
The created_time of the table
A map of table configuration. This includes fields such as delta.appendOnly, which if true indicates the table is not meant to have data deleted from it.
Get metadata from a table with the DeltaTable.metadata() method:
The schema for the table is also saved in the transaction log. It can either be retrieved in the Delta Lake form as Schema or as a PyArrow schema. The first allows you to introspect any column-level metadata stored in the schema, while the latter represents the schema the table will be loaded into.
Use DeltaTable.schema to retrieve the delta lake schema:
Depending on what system wrote the table, the delta table may have provenance information describing what operations were performed on the table, when, and by whom. This information is retained for 30 days by default, unless otherwise specified by the table configuration delta.logRetentionDuration.
Note
This information is not written by all writers and different writers may use different schemas to encode the actions. For Spark\\'s format, see: https://docs.delta.io/latest/delta-utility.html#history-schema
To view the available history, use DeltaTable.history:
The active state for a delta table is determined by the Add actions, which provide the list of files that are part of the table and metadata about them, such as creation time, size, and statistics. You can get a data frame of the add actions data using DeltaTable.get_add_actions:
The deltalake project can be installed via pip for Python or Cargo for Rust.
"},{"location":"usage/installation/#install-delta-lake-for-python","title":"Install Delta Lake for Python","text":"
With pip:
pip install deltalake\n
With Conda:
conda install -c conda-forge deltalake\n
"},{"location":"usage/installation/#install-delta-lake-for-rust","title":"Install Delta Lake for Rust","text":"
With Cargo:
cargo add deltalake\n
"},{"location":"usage/installation/#run-delta-lake-and-pandas-in-a-jupyter-notebook","title":"Run Delta Lake and pandas in a Jupyter Notebook","text":"
You can easily run Delta Lake and pandas in a Jupyter notebook.
Create an environment file with the dependencies as follows:
Depending on your storage backend, you could use the storage_options parameter to provide some configuration. Configuration is defined for specific backends - s3 options, azure options, gcs options.
The configuration can also be provided via the environment, and the basic service provider is derived from the URL being used. We try to support many of the well-known formats to identify basic service properties.
S3:
s3://\\<bucket>/\\<path>
s3a://\\<bucket>/\\<path>
Azure:
az://\\<container>/\\<path>
adl://\\<container>/\\<path>
abfs://\\<container>/\\<path>
GCS:
gs://\\<bucket>/\\<path>
Alternatively, if you have a data catalog you can load it by reference to a database and table name. Currently only AWS Glue is supported.
For AWS Glue catalog, use AWS environment variables to authenticate.
While delta always needs its internal storage backend to work and be properly configured, in order to manage the delta log, it may sometime be advantageous - and is common practice in the arrow world - to customize the storage interface used for reading the bulk data.
deltalake will work with any storage compliant with pyarrow.fs.FileSystem, however the root of the filesystem has to be adjusted to point at the root of the Delta table. We can achieve this by wrapping the custom filesystem into a pyarrow.fs.SubTreeFileSystem.
Previous table versions may not exist if they have been vacuumed, in which case an exception will be thrown. See Vacuuming tables for more information.
Vacuuming a table will delete any files that have been marked for deletion. This may make some past versions of a table invalid, so this can break time travel. However, it will save storage space. Vacuum will retain files in a certain window, by default one week, so time travel will still work in shorter ranges.
Delta tables usually don't delete old files automatically, so vacuuming regularly is considered good practice, unless the table is only appended to.
Use DeltaTable.vacuum to perform the vacuum operation. Note that to prevent accidental deletion, the function performs a dry-run by default: it will only list the files to be deleted. Pass dry_run=False to actually delete files.
>>> dt = DeltaTable(\"../rust/tests/data/simple_table\")\n>>> dt.vacuum()\n['../rust/tests/data/simple_table/part-00006-46f2ff20-eb5d-4dda-8498-7bfb2940713b-c000.snappy.parquet',\n '../rust/tests/data/simple_table/part-00190-8ac0ae67-fb1d-461d-a3d3-8dc112766ff5-c000.snappy.parquet',\n '../rust/tests/data/simple_table/part-00164-bf40481c-4afd-4c02-befa-90f056c2d77a-c000.snappy.parquet',\n ...]\n>>> dt.vacuum(dry_run=False) # Don't run this unless you are sure!\n
Delta tables can be queried in several ways. By loading as Arrow data or an Arrow dataset, they can be used by compatible engines such as Pandas and DuckDB. By passing on the list of files, they can be loaded into other engines such as Dask.
Delta tables are often larger than can fit into memory on a single computer, so this module provides ways to read only the parts of the data you need. Partition filters allow you to skip reading files that are part of irrelevant partitions. Only loading the columns required also saves memory. Finally, some methods allow reading tables batch-by-batch, allowing you to process the whole table while only having a portion loaded at any given time.
To load into Pandas or a PyArrow table use the DeltaTable.to_pandas and DeltaTable.to_pyarrow_table methods, respectively. Both of these support filtering partitions and selecting particular columns.
Converting to a PyArrow Dataset allows you to filter on columns other than partition columns and load the result as a stream of batches rather than a single table. Convert to a dataset using DeltaTable.to_pyarrow_dataset. Filters applied to datasets will use the partition values and file statistics from the Delta transaction log and push down any other filters to the scanning operation.
PyArrow datasets may also be passed to compatible query engines, such as DuckDB
>>> import duckdb\n>>> ex_data = duckdb.arrow(dataset)\n>>> ex_data.filter(\"year = 2021 and value > 4\").project(\"value\")\n---------------------\n-- Expression Tree --\n---------------------\nProjection [value]\n Filter [year=2021 AND value>4]\n arrow_scan(140409099470144, 4828104688, 1000000)\n\n---------------------\n-- Result Columns --\n---------------------\n- value (VARCHAR)\n\n---------------------\n-- Result Preview --\n---------------------\nvalue\nVARCHAR\n[ Rows: 3]\n6\n7\n5\n
Finally, you can always pass the list of file paths to an engine. For example, you can pass them to dask.dataframe.read_parquet:
For overwrites and appends, use write_deltalake. If the table does not already exist, it will be created. The data parameter will accept a Pandas DataFrame, a PyArrow Table, or an iterator of PyArrow Record Batches.
Note: write_deltalake accepts a Pandas DataFrame, but will convert it to a Arrow table before writing. See caveats in pyarrow:python/pandas.
By default, writes create a new table and error if it already exists. This is controlled by the mode parameter, which mirrors the behavior of Spark's pyspark.sql.DataFrameWriter.saveAsTable DataFrame method. To overwrite pass in mode='overwrite' and to append pass in mode='append':
write_deltalake will raise ValueError if the schema of the data passed to it differs from the existing table's schema. If you wish to alter the schema as part of an overwrite pass in overwrite_schema=True.
"},{"location":"usage/writing-delta-tables/#overwriting-a-partition","title":"Overwriting a partition","text":"
You can overwrite a specific partition by using mode=\"overwrite\" together with partition_filters. This will remove all files within the matching partition and insert your data as new files. This can only be done on one partition at a time. All of the input data must belong to that partition or else the method will raise an error.
This method could also be used to insert a new partition if one doesn't already exist, making this operation idempotent.
"},{"location":"usage/optimize/delta-lake-z-order/","title":"Delta Lake Z Order","text":"
This section explains how to Z Order a Delta table.
Z Ordering colocates similar data in the same files, which allows for better file skipping and faster queries.
Suppose you have a table with first_name, age, and country columns.
If you Z Order the data by the country column, then individuals from the same country will be stored in the same files. When you subquently query the data for individuals from a given country, it will execute faster because more data can be skipped.
"},{"location":"usage/optimize/small-file-compaction-with-optimize/","title":"Delta Lake small file compaction with optimize","text":"
This post shows you how to perform small file compaction with using the optimize method. This was added to the DeltaTable class in version 0.9.0. This command rearranges the small files into larger files which will reduce the number of files and speed up queries.
This is very helpful for workloads that append frequently. For example, if you have a table that is appended to every 10 minutes, after a year you will have 52,560 files in the table. If the table is partitioned by another dimension, you will have 52,560 files per partition; with just 100 unique values that's millions of files. By running optimize periodically, you can reduce the number of files in the table to a more manageable number.
Typically, you will run optimize less frequently than you append data. If possible, you might run optimize once you know you have finished writing to a particular partition. For example, on a table partitioned by date, you might append data every 10 minutes, but only run optimize once a day at the end of the day. This will ensure you don't need to compact the same data twice.
This section will also teach you about how to use vacuum to physically remove files from storage that are no longer needed. You\u2019ll often want vacuum after running optimize to remove the small files from storage once they\u2019ve been compacted into larger files.
Let\u2019s start with an example to explain these key concepts. All the code covered in this post is stored in this notebook in case you\u2019d like to follow along.
"},{"location":"usage/optimize/small-file-compaction-with-optimize/#create-a-delta-table-with-small-files","title":"Create a Delta table with small files","text":"
Let\u2019s start by creating a Delta table with a lot of small files so we can demonstrate the usefulness of the optimize command.
Start by writing a function that generates on thousand rows of random data given a timestamp.
def record_observations(date: datetime) -> pa.Table:\n \"\"\"Pulls data for a certain datetime\"\"\"\n nrows = 1000\n return pa.table(\n {\n \"date\": pa.array([date.date()] * nrows),\n \"timestamp\": pa.array([date] * nrows),\n \"value\": pc.random(nrows),\n }\n )\n
Let\u2019s run this function and observe the output:
Let\u2019s write 100 hours worth of data to the Delta table.
# Every hour starting at midnight on 2021-01-01\nhours_iter = (datetime(2021, 1, 1) + timedelta(hours=i) for i in itertools.count())\n\n# Write 100 hours worth of data\nfor timestamp in itertools.islice(hours_iter, 100):\n write_deltalake(\n \"observation_data\",\n record_observations(timestamp),\n partition_by=[\"date\"],\n mode=\"append\",\n )\n
This data was appended to the Delta table in 100 separate transactions, so the table will contain 100 transaction log entries and 100 data files. You can see the number of files with the files() method.
Each of these Parquet files are tiny - they\u2019re only 10 KB. Let\u2019s see how to compact these tiny files into larger files, which is more efficient for data queries.
"},{"location":"usage/optimize/small-file-compaction-with-optimize/#compact-small-files-in-the-delta-table-with-optimize","title":"Compact small files in the Delta table with optimize","text":"
Let\u2019s run the optimize command to compact the existing small files into larger files:
The optimize operation has added 5 new files and marked 100 exisitng files for removal (this is also known as \u201ctombstoning\u201d files). It has compacted the 100 tiny files into 5 larger files.
Let\u2019s append some more data to the Delta table and see how we can selectively run optimize on the new data that\u2019s added.
"},{"location":"usage/optimize/small-file-compaction-with-optimize/#handling-incremental-updates-with-optimize","title":"Handling incremental updates with optimize","text":"
Let\u2019s append another 24 hours of data to the Delta table:
for timestamp in itertools.islice(hours_iter, 24):\n write_deltalake(\n dt,\n record_observations(timestamp),\n partition_by=[\"date\"],\n mode=\"append\",\n )\n
We can use get_add_actions() to introspect the table state. We can see that 2021-01-06 has only a few hours of data so far, so we don't want to optimize that yet. But 2021-01-05 has all 24 hours of data, so it's ready to be optimized.
This optimize operation tombstones 21 small data files and adds one file with all the existing data properly condensed. Let\u2019s take a look a portion of the _delta_log/00000000000000000125.json file, which is the transaction log entry that corresponds with this incremental optimize command.
The trasaction log indicates that many files have been tombstoned and one file is added, as expected.
The Delta Lake optimize command \u201cremoves\u201d data by marking the data files as removed in the transaction log. The optimize command doesn\u2019t physically delete the Parquet file from storage. Optimize performs a \u201clogical remove\u201d not a \u201cphysical remove\u201d.
Delta Lake uses logical operations so you can time travel back to earlier versions of your data. You can vacuum your Delta table to physically remove Parquet files from storage if you don\u2019t need to time travel and don\u2019t want to pay to store the tombstoned files.
"},{"location":"usage/optimize/small-file-compaction-with-optimize/#vacuuming-after-optimizing","title":"Vacuuming after optimizing","text":"
The vacuum command deletes all files from storage that are marked for removal in the transaction log and older than the retention period which is 7 days by default.
It\u2019s normally a good idea to have a retention period of at least 7 days. For purposes of this example, we will set the retention period to zero, just so you can see how the files get removed from storage. Adjusting the retention period in this manner isn\u2019t recommended for production use cases.
All the partitions only contain a single file now, except for the date=2021-01-06 partition that has not been compacted yet.
An entire partition won\u2019t necessarily get compacted to a single data file when optimize is run. Each partition has data files that are condensed to the target file size.
"},{"location":"usage/optimize/small-file-compaction-with-optimize/#what-causes-the-small-file-problem","title":"What causes the small file problem?","text":"
Delta tables can accumulate small files for a variety of reasons:
User error: users can accidentally write files that are too small. Users should sometimes repartition in memory before writing to disk to avoid appending files that are too small.
Frequent appends: systems that append more often tend to append more smaller files. A pipeline that appends every minute will generally generate ten times as many small files compared to a system that appends every ten minutes.
Appending to partitioned data lakes with high cardinality columns can also cause small files. If you append every hour to a table that\u2019s partitioned on a column with 1,000 distinct values, then every append could create 1,000 new files. Partitioning by date avoids this problem because the data isn\u2019t split up across partitions in this manner.
This page showed you how to create a Delta table with many small files, compact the small files into larger files with optimize, and remove the tombstoned files from storage with vacuum.
You also learned about how to incrementally optimize partitioned Delta tables, so you only compact newly added data.
An excessive number of small files slows down Delta table queries, so periodic compaction is important. Make sure to properly maintain your Delta tables, so performance does not degrade over time.
"}]}
\ No newline at end of file
+{"config":{"lang":["en"],"separator":"[\\s\\-]+","pipeline":["stopWordFilter"]},"docs":[{"location":"","title":"The deltalake package","text":"
This is the documentation for the native Rust/Python implementation of Delta Lake. It is based on the delta-rs Rust library and requires no Spark or JVM dependencies. For the PySpark implementation, see delta-spark instead.
This module provides the capability to read, write, and manage Delta Lake tables with Python or Rust without Spark or Java. It uses Apache Arrow under the hood, so is compatible with other Arrow-native or integrated libraries such as pandas, DuckDB, and Polars.
\"Rust deltalake\" refers to the Rust API of delta-rs (no Spark dependency)
\"Python deltalake\" refers to the Python API of delta-rs (no Spark dependency)
\"Delta Spark\" refers to the Scala impementation of the Delta Lake transaction log protocol. This depends on Spark and Java.
"},{"location":"#why-implement-the-delta-lake-transaction-log-protocol-in-rust-and-scala","title":"Why implement the Delta Lake transaction log protocol in Rust and Scala?","text":"
Delta Spark depends on Java and Spark, which is fine for many use cases, but not all Delta Lake users want to depend on these libraries. delta-rs allows using Delta Lake in Rust or other native projects when using a JVM is often not an option.
Python deltalake lets you query Delta tables without depending on Java/Scala.
Suppose you want to query a Delta table with pandas on your local machine. Python deltalake makes it easy to query the table with a simple pip install command - no need to install Java.
Refers to the Databricks Unity Catalog <https://docs.databricks.com/data-governance/unity-catalog/index.html>_
","boost":2},{"location":"api/delta_writer/","title":"Writer","text":"","boost":10},{"location":"api/delta_writer/#write-to-delta-tables","title":"Write to Delta Tables","text":"","boost":10},{"location":"api/delta_writer/#deltalake.write_deltalake","title":"deltalake.write_deltalake","text":"
If the table does not already exist, it will be created.
This function only supports writer protocol version 2 currently. When attempting to write to an existing table with a higher min_writer_version, this function will throw DeltaProtocolError.
Note that this function does NOT register this table in a data catalog.
A locking mechanism is needed to prevent unsafe concurrent writes to a delta lake directory when writing to S3. DynamoDB is the only available locking provider at the moment in delta-rs. To enable DynamoDB as the locking provider, you need to set the AWS_S3_LOCKING_PROVIDER to 'dynamodb' as a storage_option or as an environment variable.
Additionally, you must create a DynamoDB table with the name 'delta_rs_lock_table' so that it can be automatically discovered by delta-rs. Alternatively, you can use a table name of your choice, but you must set the DYNAMO_LOCK_TABLE_NAME variable to match your chosen table name. The required schema for the DynamoDB table is as follows:
Data to write. If passing iterable, the schema must also be given.
required schemaOptional[Union[Schema, Schema]]
Optional schema to write.
Nonepartition_byOptional[Union[List[str], str]]
List of columns to partition the table by. Only required when creating a new table.
NonefilesystemOptional[FileSystem]
Optional filesystem to pass to PyArrow. If not provided will be inferred from uri. The file system has to be rooted in the table root. Use the pyarrow.fs.SubTreeFileSystem, to adopt the root of pyarrow file systems.
How to handle existing data. Default is to error if table already exists. If 'append', will add new data. If 'overwrite', will replace table with new data. If 'ignore', will not write anything if table already exists.
Optional write options for Parquet (ParquetFileWriteOptions). Can be provided with defaults using ParquetFileWriteOptions().make_write_options(). Please refer to https://github.com/apache/arrow/blob/master/python/pyarrow/_dataset_parquet.pyx#L492-L533 for the list of available options. Only used in pyarrow engine.
Nonemax_partitionsOptional[int]
the maximum number of partitions that will be used. Only used in pyarrow engine.
Nonemax_open_filesint
Limits the maximum number of files that can be left open while writing. If an attempt is made to open too many files then the least recently used file will be closed. If this setting is set too low you may end up fragmenting your data into many small files. Only used in pyarrow engine.
1024max_rows_per_fileint
Maximum number of rows per file. If greater than 0 then this will limit how many rows are placed in any single file. Otherwise there will be no limit and one file will be created in each output directory unless files need to be closed to respect max_open_files min_rows_per_group: Minimum number of rows per group. When the value is set, the dataset writer will batch incoming data and only write the row groups to the disk when sufficient rows have accumulated. Only used in pyarrow engine.
10 * 1024 * 1024max_rows_per_groupint
Maximum number of rows per group. If the value is set, then the dataset writer may split up large incoming batches into multiple row groups. If this value is set, then min_rows_per_group should also be set.
A Writer Properties instance for the Rust parquet writer.
Create a Writer Properties instance for the Rust parquet writer, see options https://arrow.apache.org/rust/parquet/file/properties/struct.WriterProperties.html:
Parameters:
Name Type Description Default data_page_size_limitOptional[int]
Limit DataPage size to this in bytes.
Nonedictionary_page_size_limitOptional[int]
Limit the size of each DataPage to store dicts to this amount in bytes.
Nonedata_page_row_count_limitOptional[int]
Limit the number of rows in each DataPage.
Nonewrite_batch_sizeOptional[int]
Splits internally to smaller batch size.
Nonemax_row_group_sizeOptional[int]
Max number of rows in row group.
NonecompressionOptional[str]
compression type
Nonecompression_levelOptional[int]
level of compression, only relevant for subset of compression types
None","boost":10},{"location":"api/delta_writer/#convert-to-delta-tables","title":"Convert to Delta Tables","text":"","boost":10},{"location":"api/delta_writer/#deltalake.convert_to_deltalake","title":"deltalake.convert_to_deltalake","text":"
Currently only HIVE partitioned tables are supported. Convert to delta creates a transaction log commit with add actions, and additional properties provided such as configuration, name, and description.
Parameters:
Name Type Description Default uriUnion[str, Path]
URI of a table.
required partition_byOptional[Schema]
Optional partitioning schema if table is partitioned.
Nonepartition_strategyOptional[Literal['hive']]
Optional partition strategy to read and convert
NonemodeLiteral['error', 'ignore']
How to handle existing data. Default is to error if table already exists. If 'ignore', will not convert anything if table already exists.
","boost":2},{"location":"api/schema/","title":"Schema","text":"","boost":2},{"location":"api/schema/#schema-and-field","title":"Schema and field","text":"
Schemas, fields, and data types are provided in the deltalake.schema submodule.
get schema with all variable size types (list, binary, string) as large variants (with int64 indices). This is for compatibility with systems like Polars that only support the large versions of Arrow types.
Create the Delta Table from a path with an optional version. Multiple StorageBackends are currently supported: AWS S3, Azure Data Lake Storage Gen2, Google Cloud Storage (GCS) and local URI. Depending on the storage backend used, you could provide options values using the storage_options parameter.
Parameters:
Name Type Description Default table_uriUnion[str, Path, PathLike[str]]
the path of the DeltaTable
required versionOptional[int]
version of the DeltaTable
Nonestorage_optionsOptional[Dict[str, str]]
a dictionary of the options to use for the storage backend
Nonewithout_filesbool
If True, will load table without tracking files. Some append-only applications might have no need of tracking any files. So, the DeltaTable will be loaded with a significant memory reduction.
Falselog_buffer_sizeOptional[int]
Number of files to buffer when reading the commit log. A positive integer. Setting a value greater than 1 results in concurrent calls to the storage api. This can decrease latency if there are many files in the log since the last checkpoint, but will also increase memory usage. Possible rate limits of the storage backend should also be considered for optimal performance. Defaults to 4 * number of cpus.
Delete expired log files before current version from table. The table log retention is based on the configuration.logRetentionDuration value, 30 days by default.
How to handle existing data. Default is to error if table already exists. If 'append', returns not support error if table exists. If 'overwrite', will CREATE_OR_REPLACE table. If 'ignore', will not do anything if table already exists. Defaults to \"error\".
Delete records from a Delta Table that statisfy a predicate.
When a predicate is not provided then all records are deleted from the Delta Table. Otherwise a scan of the Delta table is performed to mark any files that contain records that satisfy the predicate. Once files are determined they are rewritten without the records.
Parameters:
Name Type Description Default predicateOptional[str]
a SQL where clause. If not passed, will delete all rows.
Get the list of files as absolute URIs, including the scheme (e.g. \"s3://\").
Local files will be just plain absolute paths, without a scheme. (That is, no 'file://' prefix.)
Use the partition_filters parameter to retrieve a subset of files that match the given filters.
Parameters:
Name Type Description Default partition_filtersOptional[List[Tuple[str, str, Any]]]
the partition filters that will be used for getting the matched files
None
Returns:
Type Description List[str]
list of the .parquet files with an absolute URI referenced for the current version of the DeltaTable
Predicates are expressed in disjunctive normal form (DNF), like [(\"x\", \"=\", \"a\"), ...]. DNF allows arbitrary boolean logical combinations of single partition predicates. The innermost tuples each describe a single partition predicate. The list of inner predicates is interpreted as a conjunction (AND), forming a more selective and multiple partition predicates. Each tuple has format: (key, op, value) and compares the key with the value. The supported op are: =, !=, in, and not in. If the op is in or not in, the value must be a collection such as a list, a set or a tuple. The supported type for value is str. Use empty string '' for Null partition value.
The paths are as they are saved in the delta log, which may either be relative to the table root or absolute URIs.
Parameters:
Name Type Description Default partition_filtersOptional[List[Tuple[str, str, Any]]]
the partition filters that will be used for getting the matched files
None
Returns:
Type Description List[str]
list of the .parquet files referenced for the current version of the DeltaTable
Predicates are expressed in disjunctive normal form (DNF), like [(\"x\", \"=\", \"a\"), ...]. DNF allows arbitrary boolean logical combinations of single partition predicates. The innermost tuples each describe a single partition predicate. The list of inner predicates is interpreted as a conjunction (AND), forming a more selective and multiple partition predicates. Each tuple has format: (key, op, value) and compares the key with the value. The supported op are: =, !=, in, and not in. If the op is in or not in, the value must be a collection such as a list, a set or a tuple. The supported type for value is str. Use empty string '' for Null partition value.
Name Type Description Default data_catalogDataCatalog
the Catalog to use for getting the storage location of the Delta Table
required database_namestr
the database name inside the Data Catalog
required table_namestr
the table name inside the Data Catalog
required data_catalog_idOptional[str]
the identifier of the Data Catalog
NoneversionOptional[int]
version of the DeltaTable
Nonelog_buffer_sizeOptional[int]
Number of files to buffer when reading the commit log. A positive integer. Setting a value greater than 1 results in concurrent calls to the storage api. This can decrease latency if there are many files in the log since the last checkpoint, but will also increase memory usage. Possible rate limits of the storage backend should also be considered for optimal performance. Defaults to 4 * number of cpus.
Add actions represent the files that currently make up the table. This data is a low-level representation parsed from the transaction log.
Parameters:
Name Type Description Default flattenbool
whether to flatten the schema. Partition values columns are given the prefix partition., statistics (null_count, min, and max) are given the prefix null_count., min., and max., and tags the prefix tags.. Nested field names are concatenated with ..
False
Returns:
Type Description RecordBatch
a PyArrow RecordBatch containing the add action data.
Load/time travel a DeltaTable to a specified version number, or a timestamp version of the table. If a string is passed then the argument should be an RFC 3339 and ISO 8601 date and time string format.
Parameters:
Name Type Description Default versionUnion[int, str, datetime]
the identifier of the version of the DeltaTable to load
Time travel Delta table to the latest version that's created at or before provided datetime_string argument. The datetime_string argument should be an RFC 3339 and ISO 8601 date and time string.
Deprecated
Load_version and load_with_datetime have been combined into DeltaTable.load_as_version.
Parameters:
Name Type Description Default datetime_stringstr
the identifier of the datetime point of the DeltaTable to load
Pass the source data which you want to merge on the target delta table, providing a predicate in SQL query like format. You can also specify on what to do when the underlying data types do not match the underlying table.
Parameters:
Name Type Description Default sourceUnion[Table, RecordBatch, RecordBatchReader, Dataset, DataFrame]
source data
required predicatestr
SQL like predicate on how to merge
required source_aliasOptional[str]
Alias for the source table
Nonetarget_aliasOptional[str]
Alias for the target table
Noneerror_on_type_mismatchbool
specify if merge will return error if data types are mismatching :default = True
Repair the Delta Table by auditing active files that do not exist in the underlying filesystem and removes them. This can be useful when there are accidental deletions or corrupted files.
Active files are ones that have an add action in the log, but no corresponding remove action. This operation creates a new FSCK transaction containing a remove action for each of the missing or corrupted files.
Parameters:
Name Type Description Default dry_runbool
when activated, list only the files, otherwise add remove actions to transaction log. Defaults to False.
False
Returns: The metrics from repair (FSCK) action.
Example
from deltalake import DeltaTable\ndt = DeltaTable('TEST')\ndt.repair(dry_run=False)\n
Build a pandas dataframe using data from the DeltaTable.
Parameters:
Name Type Description Default partitionsOptional[List[Tuple[str, str, Any]]]
A list of partition filters, see help(DeltaTable.files_by_partitions) for filter syntax
NonecolumnsOptional[List[str]]
The columns to project. This can be a list of column names to include (order and duplicates will be preserved)
NonefilesystemOptional[Union[str, FileSystem]]
A concrete implementation of the Pyarrow FileSystem or a fsspec-compatible interface. If None, the first file path will be used to determine the right FileSystem
NonefiltersOptional[FilterType]
A disjunctive normal form (DNF) predicate for filtering rows. If you pass a filter you do not need to pass partitions
Build a PyArrow Dataset using data from the DeltaTable.
Parameters:
Name Type Description Default partitionsOptional[List[Tuple[str, str, Any]]]
A list of partition filters, see help(DeltaTable.files_by_partitions) for filter syntax
NonefilesystemOptional[Union[str, FileSystem]]
A concrete implementation of the Pyarrow FileSystem or a fsspec-compatible interface. If None, the first file path will be used to determine the right FileSystem
Build a PyArrow Table using data from the DeltaTable.
Parameters:
Name Type Description Default partitionsOptional[List[Tuple[str, str, Any]]]
A list of partition filters, see help(DeltaTable.files_by_partitions) for filter syntax
NonecolumnsOptional[List[str]]
The columns to project. This can be a list of column names to include (order and duplicates will be preserved)
NonefilesystemOptional[Union[str, FileSystem]]
A concrete implementation of the Pyarrow FileSystem or a fsspec-compatible interface. If None, the first file path will be used to determine the right FileSystem
NonefiltersOptional[FilterType]
A disjunctive normal form (DNF) predicate for filtering rows. If you pass a filter you do not need to pass partitions
Update a matched table row based on the rules defined by updates. If a predicate is specified, then it must evaluate to true for the row to be updated.
Parameters:
Name Type Description Default updatesDict[str, str]
a mapping of column name to update SQL expression.
Updating all source fields to target fields, source and target are required to have the same field names. If a predicate is specified, then it must evaluate to true for the row to be updated.
Parameters:
Name Type Description Default predicateOptional[str]
Update a target row that has no matches in the source based on the rules defined by updates. If a predicate is specified, then it must evaluate to true for the row to be updated.
Parameters:
Name Type Description Default updatesDict[str, str]
a mapping of column name to update SQL expression.
Insert a new row to the target table based on the rules defined by updates. If a predicate is specified, then it must evaluate to true for the new row to be inserted.
Parameters:
Name Type Description Default updatesdict
a mapping of column name to insert SQL expression.
Insert a new row to the target table, updating all source fields to target fields. Source and target are required to have the same field names. If a predicate is specified, then it must evaluate to true for the new row to be inserted.
Parameters:
Name Type Description Default predicateOptional[str]
Compacts small files to reduce the total number of files in the table.
This operation is idempotent; if run twice on the same table (assuming it has not been updated) it will do nothing the second time.
If this operation happens concurrently with any operations other than append, it will fail.
Parameters:
Name Type Description Default partition_filtersOptional[FilterType]
the partition filters that will be used for getting the matched files
Nonetarget_sizeOptional[int]
desired file size after bin-packing files, in bytes. If not provided, will attempt to read the table configuration value delta.targetFileSize. If that value isn't set, will use default value of 256MB.
Nonemax_concurrent_tasksOptional[int]
the maximum number of concurrent tasks to use for file compaction. Defaults to number of CPUs. More concurrent tasks can make compaction faster, but will also use more memory.
minimum interval in seconds or as timedeltas before a new commit is created. Interval is useful for long running executions. Set to 0 or timedelta(0), if you want a commit per partition.
Nonewriter_propertiesOptional[WriterProperties]
Pass writer properties to the Rust parquet writer.
None
Returns:
Type Description Dict[str, Any]
the metrics from optimize
Example
Use a timedelta object to specify the seconds, minutes or hours of the interval.
Reorders the data using a Z-order curve to improve data skipping.
This also performs compaction, so the same parameters as compact() apply.
Parameters:
Name Type Description Default columnsIterable[str]
the columns to use for Z-ordering. There must be at least one column. partition_filters: the partition filters that will be used for getting the matched files
required target_sizeOptional[int]
desired file size after bin-packing files, in bytes. If not provided, will attempt to read the table configuration value delta.targetFileSize. If that value isn't set, will use default value of 256MB.
Nonemax_concurrent_tasksOptional[int]
the maximum number of concurrent tasks to use for file compaction. Defaults to number of CPUs. More concurrent tasks can make compaction faster, but will also use more memory.
Nonemax_spill_sizeint
the maximum number of bytes to spill to disk. Defaults to 20GB.
minimum interval in seconds or as timedeltas before a new commit is created. Interval is useful for long running executions. Set to 0 or timedelta(0), if you want a commit per partition.
Nonewriter_propertiesOptional[WriterProperties]
Pass writer properties to the Rust parquet writer.
None
Returns:
Type Description Dict[str, Any]
the metrics from optimize
Example
Use a timedelta object to specify the seconds, minutes or hours of the interval.
The Parquet file stores the data that was written. The _delta_log directory stores metadata about the transactions. Let's inspect the _delta_log/00000000000000000000.json file.
This transaction adds a data file and marks the two exising data files for removal. Marking a file for removal in the transaction log is known as \"tombstoning the file\" or a \"logical delete\". This is different from a \"physical delete\" which actually removes the data file from storage.
"},{"location":"how-delta-lake-works/architecture-of-delta-table/#how-delta-table-operations-differ-from-data-lakes","title":"How Delta table operations differ from data lakes","text":"
Data lakes consist of data files persisted in storage. They don't have a transaction log that retain metadata about the transactions.
Data lakes perform transactions differently than Delta tables.
When you perform an overwrite tranasction with a Delta table, you logically delete the exiting data without physically removing it.
Data lakes don't support logical deletes, so you have to physically delete the data from storage.
Logical data operations are safer because they can be rolled back if they don't complete successfully. Physically removing data from storage can be dangerous, especially if it's before a transaction is complete.
We're now ready to look into Delta Lake ACID transactions in more detail.
"},{"location":"integrations/delta-lake-arrow/","title":"Delta Lake Arrow Integrations","text":"
Delta Lake tables can be exposed as Arrow tables and Arrow datasets, which allows for interoperability with a variety of query engines.
This page shows you how to convert Delta tables to Arrow data structures and teaches you the difference between Arrow tables and Arrow datasets. Tables are \"eager\" and datasets are \"lazy\", which has important performance implications, keep reading to learn more!
"},{"location":"integrations/delta-lake-arrow/#delta-lake-to-arrow-dataset","title":"Delta Lake to Arrow Dataset","text":"
Delta tables can easily be exposed as Arrow datasets. This makes it easy for any query engine that can read Arrow datasets to read a Delta table.
Let's take a look at the h2o groupby dataset that contains 9 columns of data. Here are three representative rows of data:
Arrow datasets allow for the predicates to get pushed down to the query engine, so the query is executed quickly.
"},{"location":"integrations/delta-lake-arrow/#delta-lake-to-arrow-table","title":"Delta Lake to Arrow Table","text":"
You can also run the same query with DuckDB on an Arrow table:
quack = duckdb.arrow(table.to_pyarrow_table())\nquack.filter(\"id1 = 'id016' and v2 > 10\")\n
This returns the same result, but it runs slower.
"},{"location":"integrations/delta-lake-arrow/#difference-between-arrow-dataset-and-arrow-table","title":"Difference between Arrow Dataset and Arrow Table","text":"
Arrow Datasets are lazy and allow for full predicate pushdown unlike Arrow tables which are eagerly loaded into memory.
The previous DuckDB queries were run on a 1 billion row dataset that's roughly 50 GB when stored as an uncompressed CSV file. Here are the runtimes when the data is stored in a Delta table and the queries are executed on a 2021 Macbook M1 with 64 GB of RAM:
Arrow table: 17.1 seconds
Arrow dataset: 0.01 seconds
The query runs much faster on an Arrow dataset because the predicates can be pushed down to the query engine and lots of data can be skipped.
Arrow tables are eagerly materialized in memory and don't allow for the same amount of data skipping.
"},{"location":"integrations/delta-lake-arrow/#multiple-query-engines-can-query-arrow-datasets","title":"Multiple query engines can query Arrow Datasets","text":"
Other query engines like DataFusion can also query Arrow datasets, see the following example:
from datafusion import SessionContext\n\nctx = SessionContext()\nctx.register_dataset(\"my_dataset\", table.to_pyarrow_dataset())\nctx.sql(\"select * from my_dataset where v2 > 5\")\n
Delta tables can easily be exposed as Arrow tables/datasets.
Therefore any query engine that can read an Arrow table/dataset can also read a Delta table.
Arrow datasets allow for more predicates to be pushed down to the query engine, so they can perform better performance than Arrow tables.
"},{"location":"integrations/delta-lake-datafusion/","title":"Using Delta Lake with DataFusion","text":"
This page explains how to use Delta Lake with DataFusion.
Delta Lake offers DataFusion users better performance and more features compared to other formats like CSV or Parquet.
Delta Lake works well with the DataFusion Rust API and the DataFusion Python API. It's a great option for all DataFusion users.
Delta Lake also depends on DataFusion to implement SQL-related functionality under the hood. We will also discuss this dependency at the end of this guide in case you're interested in learning more about the symbiotic relationship between the two libraries.
"},{"location":"integrations/delta-lake-datafusion/#delta-lake-performance-benefits-for-datafusion-users","title":"Delta Lake performance benefits for DataFusion users","text":"
Let's run some DataFusion queries on a Parquet file and a Delta table with the same data to learn more about the performance benefits of Delta Lake.
Suppose you have the following dataset with 1 billion rows and 9 columns. Here are the first three rows of data:
ctx.sql(\"select id1, sum(v1) as v1 from my_delta_table where id1='id096' group by id1\")\n
That query takes 2.8 seconds to execute.
Let's register the same dataset as a Parquet table, run the same query, and compare the runtime difference.
Register the Parquet table and run the query:
path = \"G1_1e9_1e2_0_0.parquet\"\nctx.register_parquet(\"my_parquet_table\", path)\nctx.sql(\"select id1, sum(v1) as v1 from my_parquet_table where id1='id096' group by id1\")\n
This query takes 5.3 seconds to run.
Parquet stores data in row groups and DataFusion can intelligently skip row groups that don't contain relevant data, so the query is faster than a file format like CSV which doesn't support row group skipping.
Delta Lake stores file-level metadata information in the transaction log, so it can skip entire files when queries are executed. Delta Lake can skip entire files and then skip row groups within the individual files. This makes Delta Lake even faster than Parquet files, especially for larger datasets spread across many files.
"},{"location":"integrations/delta-lake-datafusion/#delta-lake-features-for-datafusion-users","title":"Delta Lake features for DataFusion users","text":"
Delta Lake also provides other features that are useful for DataFusion users like ACID transactions, concurrency protection, time travel, versioned data, and more.
"},{"location":"integrations/delta-lake-datafusion/#why-delta-lake-depends-on-datafusion","title":"Why Delta Lake depends on DataFusion","text":"
Delta Lake depends on DataFusion to provide some end-user features.
DataFusion is useful in providing SQL-related Delta Lake features. Some examples:
Update and merge are written in terms of SQL expressions.
Invariants and constraints are written in terms of SQL expressions.
Anytime we have to evaluate SQL, we need some sort of SQL engine. We use DataFusion for that.
Delta Lake is a great file format for DataFusion users.
Delta Lake also uses DataFusion to provide some end-user features.
DataFusion and Delta Lake have a wonderful symbiotic relationship and play very nicely with each other.
See this guide for more information on Delta Lake and PyArrow and why PyArrow Datasets are often a better option than PyArrow tables.
"},{"location":"integrations/delta-lake-pandas/","title":"Using Delta Lake with pandas","text":"
Delta Lake is a great storage system for pandas analyses. This page shows how it's easy to use Delta Lake with pandas, the unique features Delta Lake offers pandas users, and how Delta Lake can make your pandas analyses run faster.
Delta Lake is very easy to install for pandas analyses, just run pip install deltalake.
Delta Lake allows for performance optimizations, so pandas queries can run much faster than the query run on data stored in CSV or Parquet. See the following chart for the query runtime for the a Delta tables compared with CSV/Parquet.
Z Ordered Delta tables run this query much faster than when the data is stored in Parquet or CSV. Let's dive in deeper and see how Delta Lake makes pandas faster.
"},{"location":"integrations/delta-lake-pandas/#delta-lake-makes-pandas-queries-run-faster","title":"Delta Lake makes pandas queries run faster","text":"
There are a few reasons Delta Lake can make pandas queries run faster:
column pruning: only grabbing the columns relevant for a query
file skipping: only reading files with data for the query
row group skipping: only reading row groups with data for the query
Z ordering data: colocating similar data in the same files, so file skipping is more effective
Reading less data (fewer columns and/or fewer rows) is how Delta Lake makes pandas queries run faster.
Parquet allows for column pruning and row group skipping, but doesn't support file-level skipping or Z Ordering. CSV doesn't support any of these performance optimizations.
Let's take a look at a sample dataset and run a query to see the performance enhancements offered by Delta Lake.
Suppose you have a 1 billion row dataset with 9 columns, here are the first three rows of the dataset:
Parquet stores data in row groups and allows for skipping when the filters predicates are set. Run the Parquet query again with row group skipping enabled:
Here are the contents after the overwrite operation (version 2 of the Delta table):
+-------+----------+\n| num | letter |\n|-------+----------|\n| 8 | dd |\n| 9 | ee |\n+-------+----------+\n
Read in the Delta table and it will grab the latest version by default:
DeltaTable(\"tmp/some-table\").to_pandas()\n\n+-------+----------+\n| num | letter |\n|-------+----------|\n| 11 | aa |\n| 22 | bb |\n+-------+----------+\n
You can easily time travel back to version 0 of the Delta table:
DeltaTable(\"tmp/some-table\", version=0).to_pandas()\n\n+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n| 3 | c |\n+-------+----------+\n
You can also time travel to version 1 of the Delta table:
DeltaTable(\"tmp/some-table\", version=1).to_pandas()\n\n+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n| 3 | c |\n| 8 | dd |\n| 9 | ee |\n+-------+----------+\n
Time travel is a powerful feature that pandas users cannot access with CSV or Parquet.
Delta tables only allow you to append DataFrame with matching schema by default. Suppose you have a DataFrame with num and animal columns, which is different from the Delta table that has columns with num and letter columns.
Try to append this DataFrame with a mismatched schema to the existing table:
This transaction will be rejected and will return the following error message:
ValueError: Schema of data does not match table schema\nData schema:\nnum: int64\nanimal: string\n-- schema metadata --\npandas: '{\"index_columns\": [{\"kind\": \"range\", \"name\": null, \"start\": 0, \"' + 474\nTable Schema:\nnum: int64\nletter: string\n
Schema enforcement protects your table from getting corrupted by appending data with mismatched schema. Parquet and CSV don't offer schema enforcement for pandas users.
"},{"location":"integrations/delta-lake-pandas/#overwriting-schema-of-table","title":"Overwriting schema of table","text":"
You can overwrite the table contents and schema by setting the overwrite_schema option. Here's how to overwrite the table contents:
Here are the contents of the table after the values and schema have been overwritten:
+-------+----------+\n| num | animal |\n|-------+----------|\n| 5 | cat |\n| 6 | dog |\n+-------+----------+\n
"},{"location":"integrations/delta-lake-pandas/#in-memory-vs-in-storage-data-changes","title":"In-memory vs. in-storage data changes","text":"
It's important to distinguish between data stored in-memory and data stored on disk when understanding the functionality offered by Delta Lake.
pandas loads data from storage (CSV, Parquet, or Delta Lake) into in-memory DataFrames.
pandas makes it easy to modify the data in memory, say update a column value. It's not easy to update a column value in storage systems like CSV or Parquet using pandas.
Delta Lake makes it easy for pandas users to update data in storage.
"},{"location":"integrations/delta-lake-pandas/#why-delta-lake-allows-for-faster-queries","title":"Why Delta Lake allows for faster queries","text":"
Delta tables store data in many files and metadata about the files in the transaction log. Delta Lake allows for certain queries to skip entire files, which makes pandas queries run much faster.
Delta Lake provides many features that make it an excellent format for pandas analyses:
performance optimizations make pandas queries run faster
data management features make pandas analyses more reliable
advanced features allow you to perform more complex pandas analyses
Python deltalake offers pandas users a better experience compared with CSV/Parquet.
"},{"location":"integrations/delta-lake-polars/","title":"Using Delta Lake with polars","text":"
This page explains why Delta Lake is a great storage system for Polars analyses.
You will learn how to create Delta tables with Polars, how to query Delta tables with Polars, and the unique advantages Delta Lake offers the Polars community.
Here are some amazing benefits that Delta Lake provides Polars users:
time travel
ACID transactions for reliable writes
better performance with file skipping
enhanced file skipping via Z Ordering
ability to rollback mistakes
and many, many more
Let's start by showing how to use Polars with Delta Lake, explore how Delta Lake can make Polars queries run faster, and then look at all the cool features Delta Lake offers Polars users.
"},{"location":"integrations/delta-lake-polars/#creating-a-delta-lake-table-with-polars","title":"Creating a Delta Lake table with Polars","text":"
Create a Polars DataFrame and write it out to a Delta table:
import polars as pl\n\ndf = pl.DataFrame({\"x\": [1, 2, 3]})\ndf.write_delta(\"tmp/bear_delta_lake\")\n
This dataset is 50GB when stored in an uncompressed CSV file. Let's run some queries on this dataset when it's stored in different file formats with Polars.
This section will show the runtime for a query when the data is stored in CSV, Parquet, and Delta Lake and explain why Delta tables are the fastest.
Start by running a query on an uncompressed CSV file with read_csv:
This query runs in 8.3 seconds. It's much faster because Polars is optimized to skip row groups in Parquet files that don't contain data that's relevant for the query.
This query runs in 7.2 seconds. Polars can run this query faster because it can inspect the Delta transaction log and skip entire files that don't contain relevant data before performing the ordinary Parquet row group skipping.
Finally run the query on the Delta table after it has been Z Ordered by id1:
This query runs in 3.5 seconds. The query on the Z Ordered Delta table is even faster because similar data has been co-located in the same files. This allows for even greater data skipping.
Polars can leverage file skipping to query Delta tables very quickly.
"},{"location":"integrations/delta-lake-polars/#why-polars-is-fast-with-delta-lake","title":"Why Polars is fast with Delta Lake","text":"
Delta tables consist of metadata in a transaction log and data stored in Parquet files.
When Polars queries a Delta table, it starts by consulting the transaction log to understand the metadata of each file in the Delta table. This allows for Polars to quickly identify which files should be skipped by the query.
CSV files don't contain any such metadata, so file skipping isn't an option. Polars can skip Parquet files based on metadata, but it needs to open up each file and read the metadata, which is slower that grabbing the file-level metadata directly from the transaction log.
Parquet doesn't allow users to easily Z Order the data and colocate similar data in the same row groups. The Z Order optimizations are only supported in Delta tables.
Delta Lake offers Polars users with unique performance optimizations.
"},{"location":"integrations/delta-lake-polars/#other-delta-lake-features-relevant-for-polars-users","title":"Other Delta Lake features relevant for Polars users","text":"
This guide shows how Delta Lake is a great storage format for Polars analyses.
Delta Lake is easy to use, fast, and full of features that are great for Polars users.
"},{"location":"usage/","title":"Usage","text":"
A DeltaTable represents the state of a delta table at a particular version. This includes which files are currently part of the table, the schema of the table, and other metadata such as creation time.
Python Rust
DeltaTable
from deltalake import DeltaTable\n\ndt = DeltaTable(\"../rust/tests/data/delta-0.2.0\")\nprint(f\"Version: {dt.version()}\")\nprint(f\"Files: {dt.files()}\")\n
DeltaTable
let table = deltalake::open_table(\"../rust/tests/data/simple_table\").await.unwrap();\nprintln!(\"Version: {}\", table.version());\nprintln!(\"Files: {}\", table.get_files());\n
"},{"location":"usage/appending-overwriting-delta-lake-table/","title":"Appending to and overwriting a Delta Lake table","text":"
This section explains how to append to an exising Delta table and how to overwrite a Delta table.
"},{"location":"usage/appending-overwriting-delta-lake-table/#delta-lake-append-transactions","title":"Delta Lake append transactions","text":"
Suppose you have a Delta table with the following contents:
+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n| 3 | c |\n+-------+----------+\n
Here are the contents of the Delta table after the overwrite operation:
+-------+----------+\n| num | letter |\n|-------+----------|\n| 11 | aa |\n| 22 | bb |\n+-------+----------+\n
Overwriting just performs a logical delete. It doesn't physically remove the previous data from storage. Time travel back to the previous version to confirm that the old version of the table is still accessable.
dt = DeltaTable(\"tmp/some-table\", version=1)\n\n+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n| 3 | c |\n| 8 | dd |\n| 9 | ee |\n+-------+----------+\n
"},{"location":"usage/create-delta-lake-table/","title":"Creating a Delta Lake Table","text":"
This section explains how to create a Delta Lake table.
You can easily write a DataFrame to a Delta table.
from deltalake import write_deltalake\nimport pandas as pd\n\ndf = pd.DataFrame({\"num\": [1, 2, 3], \"letter\": [\"a\", \"b\", \"c\"]})\nwrite_deltalake(\"tmp/some-table\", df)\n
Here are the contents of the Delta table in storage:
+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n| 3 | c |\n+-------+----------+\n
"},{"location":"usage/deleting-rows-from-delta-lake-table/","title":"Deleting rows from a Delta Lake table","text":"
This section explains how to delete rows from a Delta Lake table.
Suppose you have the following Delta table with four rows:
+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n| 3 | c |\n| 4 | d |\n+-------+----------+\n
Here's how to delete all the rows where the num is greater than 2:
Here are the contents of the Delta table after the delete operation has been performed:
+-------+----------+\n| num | letter |\n|-------+----------|\n| 1 | a |\n| 2 | b |\n+-------+----------+\n
"},{"location":"usage/examining-table/","title":"Examining a Table","text":""},{"location":"usage/examining-table/#metadata","title":"Metadata","text":"
The delta log maintains basic metadata about a table, including:
A unique id
A name, if provided
A description, if provided
The list of partitionColumns.
The created_time of the table
A map of table configuration. This includes fields such as delta.appendOnly, which if true indicates the table is not meant to have data deleted from it.
Get metadata from a table with the DeltaTable.metadata() method:
The schema for the table is also saved in the transaction log. It can either be retrieved in the Delta Lake form as Schema or as a PyArrow schema. The first allows you to introspect any column-level metadata stored in the schema, while the latter represents the schema the table will be loaded into.
Use DeltaTable.schema to retrieve the delta lake schema:
Depending on what system wrote the table, the delta table may have provenance information describing what operations were performed on the table, when, and by whom. This information is retained for 30 days by default, unless otherwise specified by the table configuration delta.logRetentionDuration.
Note
This information is not written by all writers and different writers may use different schemas to encode the actions. For Spark\\'s format, see: https://docs.delta.io/latest/delta-utility.html#history-schema
To view the available history, use DeltaTable.history:
The active state for a delta table is determined by the Add actions, which provide the list of files that are part of the table and metadata about them, such as creation time, size, and statistics. You can get a data frame of the add actions data using DeltaTable.get_add_actions:
The deltalake project can be installed via pip for Python or Cargo for Rust.
"},{"location":"usage/installation/#install-delta-lake-for-python","title":"Install Delta Lake for Python","text":"
With pip:
pip install deltalake\n
With Conda:
conda install -c conda-forge deltalake\n
"},{"location":"usage/installation/#install-delta-lake-for-rust","title":"Install Delta Lake for Rust","text":"
With Cargo:
cargo add deltalake\n
"},{"location":"usage/installation/#run-delta-lake-and-pandas-in-a-jupyter-notebook","title":"Run Delta Lake and pandas in a Jupyter Notebook","text":"
You can easily run Delta Lake and pandas in a Jupyter notebook.
Create an environment file with the dependencies as follows:
Depending on your storage backend, you could use the storage_options parameter to provide some configuration. Configuration is defined for specific backends - s3 options, azure options, gcs options.
The configuration can also be provided via the environment, and the basic service provider is derived from the URL being used. We try to support many of the well-known formats to identify basic service properties.
S3:
s3://\\<bucket>/\\<path>
s3a://\\<bucket>/\\<path>
Azure:
az://\\<container>/\\<path>
adl://\\<container>/\\<path>
abfs://\\<container>/\\<path>
GCS:
gs://\\<bucket>/\\<path>
Alternatively, if you have a data catalog you can load it by reference to a database and table name. Currently only AWS Glue is supported.
For AWS Glue catalog, use AWS environment variables to authenticate.
While delta always needs its internal storage backend to work and be properly configured, in order to manage the delta log, it may sometime be advantageous - and is common practice in the arrow world - to customize the storage interface used for reading the bulk data.
deltalake will work with any storage compliant with pyarrow.fs.FileSystem, however the root of the filesystem has to be adjusted to point at the root of the Delta table. We can achieve this by wrapping the custom filesystem into a pyarrow.fs.SubTreeFileSystem.
Previous table versions may not exist if they have been vacuumed, in which case an exception will be thrown. See Vacuuming tables for more information.
Vacuuming a table will delete any files that have been marked for deletion. This may make some past versions of a table invalid, so this can break time travel. However, it will save storage space. Vacuum will retain files in a certain window, by default one week, so time travel will still work in shorter ranges.
Delta tables usually don't delete old files automatically, so vacuuming regularly is considered good practice, unless the table is only appended to.
Use DeltaTable.vacuum to perform the vacuum operation. Note that to prevent accidental deletion, the function performs a dry-run by default: it will only list the files to be deleted. Pass dry_run=False to actually delete files.
>>> dt = DeltaTable(\"../rust/tests/data/simple_table\")\n>>> dt.vacuum()\n['../rust/tests/data/simple_table/part-00006-46f2ff20-eb5d-4dda-8498-7bfb2940713b-c000.snappy.parquet',\n '../rust/tests/data/simple_table/part-00190-8ac0ae67-fb1d-461d-a3d3-8dc112766ff5-c000.snappy.parquet',\n '../rust/tests/data/simple_table/part-00164-bf40481c-4afd-4c02-befa-90f056c2d77a-c000.snappy.parquet',\n ...]\n>>> dt.vacuum(dry_run=False) # Don't run this unless you are sure!\n
Delta tables can be queried in several ways. By loading as Arrow data or an Arrow dataset, they can be used by compatible engines such as Pandas and DuckDB. By passing on the list of files, they can be loaded into other engines such as Dask.
Delta tables are often larger than can fit into memory on a single computer, so this module provides ways to read only the parts of the data you need. Partition filters allow you to skip reading files that are part of irrelevant partitions. Only loading the columns required also saves memory. Finally, some methods allow reading tables batch-by-batch, allowing you to process the whole table while only having a portion loaded at any given time.
To load into Pandas or a PyArrow table use the DeltaTable.to_pandas and DeltaTable.to_pyarrow_table methods, respectively. Both of these support filtering partitions and selecting particular columns.
Converting to a PyArrow Dataset allows you to filter on columns other than partition columns and load the result as a stream of batches rather than a single table. Convert to a dataset using DeltaTable.to_pyarrow_dataset. Filters applied to datasets will use the partition values and file statistics from the Delta transaction log and push down any other filters to the scanning operation.
PyArrow datasets may also be passed to compatible query engines, such as DuckDB
>>> import duckdb\n>>> ex_data = duckdb.arrow(dataset)\n>>> ex_data.filter(\"year = 2021 and value > 4\").project(\"value\")\n---------------------\n-- Expression Tree --\n---------------------\nProjection [value]\n Filter [year=2021 AND value>4]\n arrow_scan(140409099470144, 4828104688, 1000000)\n\n---------------------\n-- Result Columns --\n---------------------\n- value (VARCHAR)\n\n---------------------\n-- Result Preview --\n---------------------\nvalue\nVARCHAR\n[ Rows: 3]\n6\n7\n5\n
Finally, you can always pass the list of file paths to an engine. For example, you can pass them to dask.dataframe.read_parquet:
For overwrites and appends, use write_deltalake. If the table does not already exist, it will be created. The data parameter will accept a Pandas DataFrame, a PyArrow Table, or an iterator of PyArrow Record Batches.
Note: write_deltalake accepts a Pandas DataFrame, but will convert it to a Arrow table before writing. See caveats in pyarrow:python/pandas.
By default, writes create a new table and error if it already exists. This is controlled by the mode parameter, which mirrors the behavior of Spark's pyspark.sql.DataFrameWriter.saveAsTable DataFrame method. To overwrite pass in mode='overwrite' and to append pass in mode='append':
write_deltalake will raise ValueError if the schema of the data passed to it differs from the existing table's schema. If you wish to alter the schema as part of an overwrite pass in overwrite_schema=True.
"},{"location":"usage/writing-delta-tables/#overwriting-a-partition","title":"Overwriting a partition","text":"
You can overwrite a specific partition by using mode=\"overwrite\" together with partition_filters. This will remove all files within the matching partition and insert your data as new files. This can only be done on one partition at a time. All of the input data must belong to that partition or else the method will raise an error.
This method could also be used to insert a new partition if one doesn't already exist, making this operation idempotent.
"},{"location":"usage/optimize/delta-lake-z-order/","title":"Delta Lake Z Order","text":"
This section explains how to Z Order a Delta table.
Z Ordering colocates similar data in the same files, which allows for better file skipping and faster queries.
Suppose you have a table with first_name, age, and country columns.
If you Z Order the data by the country column, then individuals from the same country will be stored in the same files. When you subquently query the data for individuals from a given country, it will execute faster because more data can be skipped.
"},{"location":"usage/optimize/small-file-compaction-with-optimize/","title":"Delta Lake small file compaction with optimize","text":"
This post shows you how to perform small file compaction with using the optimize method. This was added to the DeltaTable class in version 0.9.0. This command rearranges the small files into larger files which will reduce the number of files and speed up queries.
This is very helpful for workloads that append frequently. For example, if you have a table that is appended to every 10 minutes, after a year you will have 52,560 files in the table. If the table is partitioned by another dimension, you will have 52,560 files per partition; with just 100 unique values that's millions of files. By running optimize periodically, you can reduce the number of files in the table to a more manageable number.
Typically, you will run optimize less frequently than you append data. If possible, you might run optimize once you know you have finished writing to a particular partition. For example, on a table partitioned by date, you might append data every 10 minutes, but only run optimize once a day at the end of the day. This will ensure you don't need to compact the same data twice.
This section will also teach you about how to use vacuum to physically remove files from storage that are no longer needed. You\u2019ll often want vacuum after running optimize to remove the small files from storage once they\u2019ve been compacted into larger files.
Let\u2019s start with an example to explain these key concepts. All the code covered in this post is stored in this notebook in case you\u2019d like to follow along.
"},{"location":"usage/optimize/small-file-compaction-with-optimize/#create-a-delta-table-with-small-files","title":"Create a Delta table with small files","text":"
Let\u2019s start by creating a Delta table with a lot of small files so we can demonstrate the usefulness of the optimize command.
Start by writing a function that generates on thousand rows of random data given a timestamp.
def record_observations(date: datetime) -> pa.Table:\n \"\"\"Pulls data for a certain datetime\"\"\"\n nrows = 1000\n return pa.table(\n {\n \"date\": pa.array([date.date()] * nrows),\n \"timestamp\": pa.array([date] * nrows),\n \"value\": pc.random(nrows),\n }\n )\n
Let\u2019s run this function and observe the output:
Let\u2019s write 100 hours worth of data to the Delta table.
# Every hour starting at midnight on 2021-01-01\nhours_iter = (datetime(2021, 1, 1) + timedelta(hours=i) for i in itertools.count())\n\n# Write 100 hours worth of data\nfor timestamp in itertools.islice(hours_iter, 100):\n write_deltalake(\n \"observation_data\",\n record_observations(timestamp),\n partition_by=[\"date\"],\n mode=\"append\",\n )\n
This data was appended to the Delta table in 100 separate transactions, so the table will contain 100 transaction log entries and 100 data files. You can see the number of files with the files() method.
Each of these Parquet files are tiny - they\u2019re only 10 KB. Let\u2019s see how to compact these tiny files into larger files, which is more efficient for data queries.
"},{"location":"usage/optimize/small-file-compaction-with-optimize/#compact-small-files-in-the-delta-table-with-optimize","title":"Compact small files in the Delta table with optimize","text":"
Let\u2019s run the optimize command to compact the existing small files into larger files:
The optimize operation has added 5 new files and marked 100 exisitng files for removal (this is also known as \u201ctombstoning\u201d files). It has compacted the 100 tiny files into 5 larger files.
Let\u2019s append some more data to the Delta table and see how we can selectively run optimize on the new data that\u2019s added.
"},{"location":"usage/optimize/small-file-compaction-with-optimize/#handling-incremental-updates-with-optimize","title":"Handling incremental updates with optimize","text":"
Let\u2019s append another 24 hours of data to the Delta table:
for timestamp in itertools.islice(hours_iter, 24):\n write_deltalake(\n dt,\n record_observations(timestamp),\n partition_by=[\"date\"],\n mode=\"append\",\n )\n
We can use get_add_actions() to introspect the table state. We can see that 2021-01-06 has only a few hours of data so far, so we don't want to optimize that yet. But 2021-01-05 has all 24 hours of data, so it's ready to be optimized.
This optimize operation tombstones 21 small data files and adds one file with all the existing data properly condensed. Let\u2019s take a look a portion of the _delta_log/00000000000000000125.json file, which is the transaction log entry that corresponds with this incremental optimize command.
The trasaction log indicates that many files have been tombstoned and one file is added, as expected.
The Delta Lake optimize command \u201cremoves\u201d data by marking the data files as removed in the transaction log. The optimize command doesn\u2019t physically delete the Parquet file from storage. Optimize performs a \u201clogical remove\u201d not a \u201cphysical remove\u201d.
Delta Lake uses logical operations so you can time travel back to earlier versions of your data. You can vacuum your Delta table to physically remove Parquet files from storage if you don\u2019t need to time travel and don\u2019t want to pay to store the tombstoned files.
"},{"location":"usage/optimize/small-file-compaction-with-optimize/#vacuuming-after-optimizing","title":"Vacuuming after optimizing","text":"
The vacuum command deletes all files from storage that are marked for removal in the transaction log and older than the retention period which is 7 days by default.
It\u2019s normally a good idea to have a retention period of at least 7 days. For purposes of this example, we will set the retention period to zero, just so you can see how the files get removed from storage. Adjusting the retention period in this manner isn\u2019t recommended for production use cases.
All the partitions only contain a single file now, except for the date=2021-01-06 partition that has not been compacted yet.
An entire partition won\u2019t necessarily get compacted to a single data file when optimize is run. Each partition has data files that are condensed to the target file size.
"},{"location":"usage/optimize/small-file-compaction-with-optimize/#what-causes-the-small-file-problem","title":"What causes the small file problem?","text":"
Delta tables can accumulate small files for a variety of reasons:
User error: users can accidentally write files that are too small. Users should sometimes repartition in memory before writing to disk to avoid appending files that are too small.
Frequent appends: systems that append more often tend to append more smaller files. A pipeline that appends every minute will generally generate ten times as many small files compared to a system that appends every ten minutes.
Appending to partitioned data lakes with high cardinality columns can also cause small files. If you append every hour to a table that\u2019s partitioned on a column with 1,000 distinct values, then every append could create 1,000 new files. Partitioning by date avoids this problem because the data isn\u2019t split up across partitions in this manner.
This page showed you how to create a Delta table with many small files, compact the small files into larger files with optimize, and remove the tombstoned files from storage with vacuum.
You also learned about how to incrementally optimize partitioned Delta tables, so you only compact newly added data.
An excessive number of small files slows down Delta table queries, so periodic compaction is important. Make sure to properly maintain your Delta tables, so performance does not degrade over time.