Please enable JavaScript.

Coggle requires JavaScript to display documents.

Apache Iceberg - Coggle Diagram

- - - - rewriteDataFiles is the actions = compaction
    - - BinPack (default compaction strategy)
        
        Combines files only; no global sorting (will do local
        sorting within tasks)
        
        pros
        
        This offers the fastest compaction jobs.
        
        cons
        
        Data is not clustered.
      - Sort
        
        Sorts by one or more fields sequentially prior to allocating tasks (e.g., sort by field a, then within that, sort by field b)
        
        pros
        
        Faster read time after sorting the rewritten data
        
        cons
        
        The compaction is longer than the binPacking strategy
      - zOrder
        
        Sorts by multiple fields equally weighted -> grouping frequently used as filter columns together
        
        pros
        
        Faster read time for query filtering on multiple fields.
        
        cons
        
        Even more time to run the compaction job
      - filter
    - - target-file-size-bytes
      - max-concurrent-file-group-rewrites
      - max-file-group-size-bytes
      - partial-progress-enabled
      - partial-progress-max-commits
      - rewrite-job-order
    - - For Apache Parquet files, there is row group size and a file size.\ The row group size is the size of one group of rows in a Parquet file, and each file can have multiple groups. Therefore, an Iceberg table’s default configuration would allow for a 128 MB row group size and a 512 MB file size (four row groups per file).
        You’ll always want to make sure these two settings are aligned (i.e., that the row size is evenly divided by the file size).
        Fewer row groups results in a smaller file size, as there are fewer groups to have group metadata written for, while more row groups improves predicate pushdown because the row group metadata can have more fine-grained ranges, making it possible for the query engine to eliminate reading more row groups that don’t contain data relevant to the current query. Another example is that you may want to increase the file size to 1 GB per file but keep row groups to 128 MB (eight row groups per file); that way, there are fewer files to open and close. Although the types of queries you’re running often require reading most of the data, you’d prefer fewer row groups since predicate pushdown will not speed up the process of getting all the data. Row group size and file size can both be set as table properties (write.parquet.rowgroup-size-bytes and write.target-file-size-bytes, respectively), but the file size can be set for individual compaction jobs using the options settings.
    - - Depending on context it could become a nightmare to manually run all the compactions.
      - few approaches possibles
        
        Use an orchestration tool as Airflow, Dagster, Perfect, Argo or Luigi
        
        Use serverless compute engine (AWS lamda, Azure function) triggered by FS changed
        
        Setup a simple cronjob to run the compaction at periodic time
  - - - ALTER TABLE catalog.nfl_teams WRITE ORDERED BY team;
    - - INSERT INTO catalog.nfl_teams SELECT * FROM staging_table ORDER BY team
    - - CALL catalog.system.rewrite_data_files( table => 'nfl_teams', strategy => 'sort', sort_order => 'team ASC NULLS LAST, name ASC NULLS FIRST' )
    - - Considering a sorting on the Team name here for the NFL table. What if we need all the Lion team players which their name start with an A? We can then implement a 2nd layer sorting on the player name. But what if we need the player Name starting with A across all NFL teams ? The sequential sorting is in this case inefficient as A players are stretched all partition.
        This is where z-ordering can be useful.
  - - - partition 1 -> age 1-50 ; height 1 to 100cm
      - partition 2 -> age 1-50 ; height 100 to 200cm
      - partition 3 -> age 51-100 ; height 1 to 100cm
      - partition 4 -> age 51-100 ; height 100 to 200cm
      - With this partitioning, let's consider the following query: SELECT * FROM mytable WHERE age = 60 AND height = 170cm
        
        We are reducing the number of datafile scanned per 75% as the only part usefull is the number 4.
  - - - Instead of tracking partitioning relying on how files are physically laid out, Iceberg tracks the range of partition values at the
        snapshot and manifest levels
        
        Relying on built-in transforms
        
        Facilitate the query filter reducing the need to deep understand how the partitioning is physically done, you just have to know on which column the partition scheme is relying on.
        
        Don't need to add additional column for the partitioning as they are calculated on existing column with the transforms built-in functions
      - Partitioned table example:
        CREATE TABLE catalog.MyTable (...) PARTITIONED BY months(time) USING iceberg;
      - Query which transparently benefits from this partitioning scheme : SELECT * FROM MYTABLE WHERE time BETWEEN '2022-07-01 00:00:00' AND '2022-07-31
        00:00:00';
    - - As the hive partitioning is relying on folders structure, you have to rewrite the entire dataset if you want to change the partitioning scheme
      - Partitioning scheme with Iceberg is relying on metadata files, allowing the change the partitioning scheme from a snapshot to another
      - Example:
      - CREATE TABLE catalog.members (...) PARTITIONED BY years(registration_ts) USING
        iceberg;
      - ALTER TABLE catalog.members ADD PARTITION FIELD months(registration_ts)
      - :warning: When a partitioning scheme is updated, it only applies to new data written to the table going forward, so there is no need to rewrite the existing data. Also, keep in mind that any data rewritten by the rewriteDataFiles procedure will be rewritten using the new partitioning scheme, so if you want to keep older data in the old scheme, make sure to use the proper filters in your compaction jobs to not rewrite it.
  - - - Copy-On-Write (COW) -> the default strategy
        
        In this approach, new datafiles are generated with the updated or deleted rows at the writing time.
        
        But slowest updates / deletes as datafiles have to be rewritten entirely
        
        Provide the fastest reads table as there is no merge operation to do it reading time
      - Merge-On-Read (MOR)
        
        POSITION
        
        + fatest read, - write as the writer need to calculate the position
        
        EQUALLITY
        
        - higher read cost, + fatest write possible
        
        Instead of rewriting an entire datafile, you capture in a delete file the records to be updated in the existing file, with the delete file tracking which records should be ignored.
        
        When a reader reads the table, it will reconcile the datafile with the delete file.
        
        Same for an update, a new datafile is created with the updates row only and a delete file is creating to remove old rows from previous datafile
    - - 'write.update.mode'='copy-on-write'
      - 'write.delete.mode'='merge-on-read'
      - 'write.merge.mode'='copy-on-write'
  - - - Statistic can become too heavy when tables have lot of fields >100.
      - Fine tune which columns have theirs metrics tracked and which don't.
        
        Can specify the metric collection strategy in the tbls properties : write.metadata.metrics.column.col1'='none'
        
        none = don't collect any metrics
        
        counts = only collect counts (values, distinct values, null values)
        
        truncate(XX) = truncate the value on which metrics are calculated. DEFAULT truncate(16)
        
        full = metrics are calculated on the full value
    - - Rewrite only the manifests files : CALL catalog.system.rewrite_manifests("TABLE_NAME", TRUE/FALSE). The 2nd param is a boolean to desactivate spark cache avoiding OOM
    - - Default time-travel depth is 5days but we can expire snapshot manually
        
        CALL catalog.system.expire_snapshots(table => 'MyTable', snapshot_ids => ARRAY(53))
        
        table = the table to expire snapshots on
        
        older_than = expires all snapshot before a given timestamp
        
        retain_last = minimum number of snapshot to retain
        
        snapshot_ids = specific ids to expire
        
        max_concurrent_deletes = number of threads to use for deleting files
        
        stream_results = boolean true => send deleted files to the spark driver by RDD partition avoiding OOM issues on deleting large datafile
      - Failed writing operation can result in having orphans files. (Orphan files = any file not tracked by a snapshots)
        
        CALL catalog.system.remove_orphan_files(table => 'MyTable')
        
        table = the table to clean orphans on
        
        older_than = Only deletes files created before a given timestamp
        
        location = where to look for orphans files. Default is the table location
        
        dry_run = boolean if true -> return the list of what would be deleted
        
        max_concurrent_deletes = list the max number of threads for deleting files
    - - Depending on the object implementation get lot of files with the same prefix as /prefix1/fileA, /prefix1/fileB etc.. can have limitation and throttling may appear.
      - To avoid this Iceberg define the following table properties 'write.object-storage.enabled'= true which add a hash in the prefix
    - - A bloom filter is a way of knowing whether a value possibly exists in a dataset.
      - Usefull to avoid unnecessary data scans
      - :warning: It's a trade-off, the more going trough it the less the quality is. A good bloom filter needs to be balanced.
      - Iceberg provide a table properties to enable bloom-filtering on specific col with the 2 following properties :
        'write.parquet.bloom-filter-enabled.column.col1'= true,
        'write.parquet.bloom-filter-max-bytes'= 1048576
- - - - RDBMS ?
        
        Relational DataBase Management System
        
        Standard transactional data storage
        
        Best for OLTP (Online transaction processing)
        
        Examples : MySQL, Postgres, SQL Server...
        
        Optimized to interact very quickly on small operation (dealing with one or few rows of data)
      - Why it is not suitable for OLAP
        
        Transactional systems are focused on inserting, updating, and reading a small subset of rows in a table.
        
        Running transactional and analytics workloads on the same infrastructure can result in a competition for resources.
        
        Transactional workloads benefit from normalizing the data into several related tables that are joined if needed, while analytics workloads may perform better when the data is denormalized into the same table to avoid large-scale join operations.
    - - Data warehouse ?
        
        A data warehouse or OLAP database is a centralized repository that supports storing large volumes of data ingested from various sources such as operational systems, application databases, and logs
        
        Once data ingested, it's become fully managed by the data warehouse itself (it's own storage, own proprietary format, proprietary table format)
        
        pros
        
        serve as the data golden source
        
        Support querying large amount of historical data (OLAP compliant)
        
        Provide effective data gouvernance with quality and security by design
        
        Automation on the data layout (the way the data are stored)
        
        Schema enforcement
        
        cons
        
        Data cannot be consumed from other engine. It locks the data as only usable by the vendor-specific system
        
        this make hard implementing new ML workload
        
        Cost is expensive & hard to manage
        
        Mostly support only structured data
    - - Datalake ?
        
        Goal was to achieve the same as data warehouse (analytical workload on huge amount of data) but in a cost effective manner and fully open.
        
        Hadoop
        
        Open source
        
        HDFS High Distributed File System
        
        Structured and semi / unstructured data
        
        MapReduce framework
        
        Replaced now by Spark
        
        Hive (SQL motor to transform SQL queries to MR job)
        
        Over time, companies are moving out form Hadoop to use cloud object storage (ADLS, S3 ...)
        
        Easier to manage
        
        Cost effective
        
        Pros
        
        Lower cost
        
        Data stored in open format
        
        Compatible with structured / semi-structured and unstructured data
        
        Cons
        
        Performance is hard to achieve
        
        No managed indexes
        
        Required to really think / re-think the performance by optimizing at every levels. Storage, Format, Engine
        
        Too fined grained configuration
        
        required data engineering skill to configure everything
        
        Lack of ACID transactions
        
        ACID ?
        
        Challenging for use case with strong transactional integrity requirement (Financial system, sales,...)
        
        Achieving similar transactional guarantees involves implementing complex data pipelines and coordination mechanisms.
    - - Data lakehouse ?
        
        Kind of the same architecture as Datalake (decoupling storage and compute) + bringing data warehouse-like features ( ACID transaction, performance by design ...)
        
        Table format
        
        Allowing to switch from append-only datalake to the real center of the data world.
        
        Provide a metadata/abstraction layer between the storage layer and the compute layer, making them work more intelligently
        
        Reducing the need to have several copies of the same data
        
        Faster queries by using caching, cost-based optimizer, better pruning, query plan based on metadata
        
        Remain opensource as datalake
    - - Data is a primary asset from which organizations curate the information and insights needed to make critical business decisions.
      - Data is growing fast, more and more collected sources
      - Decicion based on analytical (BI)
      - Decison based on ML models
  - - - Hive was developed by Facebook in 2009, at that date it was painful to write java MapReduce job to query the data
      - Hive introduced the table data representation on datalake. Tables are managed by a dedicated service called Hive metastore service (HMS)
      - Hive took the approach of defining a table as files within a specified directory (or prefix in object storage language). Partitions are subdirectories
      - Benefits
        
        It enabled more efficient query patterns than full table scans, so techniques such as partitioning and bucketing
        
        It is file format agnostic
        
        Become the standard over the time
      - Limitations
        
        You are essentially left making swaps at the partition level to update a single file atomically
        
        There isn't a mechanism for atomically updating multiple partitions as one transaction. This opens up the possibility for end users seeing inconsistent data between transactions updating multiple partitions.
        
        Partitioning only helped if you filtered by the partition column, and someone who has a filter on the timestamp column may not intuitively know to also filter on the derived month column, leading to a full table scan since partitioning was not taken advantage of
        
        Since object storage often throttles requests against the same prefix (think of an object storage prefix as analogous to a file directory), queries on tables with large numbers of files in a single partition (so that all the files would be in one prefix) can have performance issues.
    - - In order to solve those limitations from the Hive approach, a new generation of table formats arose.
      - Modern table formats such as Apache Iceberg, Apache Hudi, and Delta Lake all took this approach of defining tables as a canonical list of files, providing metadata for engines informing which files make up the table, not which directories.
      - benefits
        
        Allow ACID transactions
        
        Better statistics/metadata collection in order to prune maximum data at querying
  - - - Apple
      - dreamio
      - AWS
      - Tencent
      - LinkedIn
      - Stripe
    - - Consistency: Update to a table across multiples partition should be done atomically so that data remain consistent to the consuming end user
      - Performance: With Hive’s file/directory listing bottleneck, query planning would take excessively long to complete before actually executing the query. The table should provide metadata and avoid excessive file listing so that not only can query planning be a faster process, but also the resulting plans can be executed more quickly since they scan only the files necessary to satisfy the query
      - Schema evolution: A table should be able to evolve its schema and partitioning scheme safely and without the need for rewriting
      - Scalability: All the goals should be able to be accomplished at petabytes scale
  - - - Manifest file
        
        A list of datafiles, containing each datafile’s location/path and key metadata about those datafiles, which allows for creating more efficient execution plans
      - Manifest list
        
        Files that define a single snapshot of the table as a list of manifest files along with stats on those manifests that allow for creating more efficient execution plans.
      - Metadata file
        
        Files that define a table’s structure, including its schema, partitioning scheme, and a listing of snapshots.
      - Catalog
        
        Tracks the table location (similar to the Hive Metastore), but instead of containing a mapping of table name -> set of directories, it contains a mapping of table name -> location of the table’s most recent metadata file
  - - - Apache Iceberg uses optimistic concurrency control
        
        Optimistic concurrency assumes transactions won't conflict and checks for conflicts only when necessary.
        
        Minimize locking to improve performance
        
        Pessimistic concurrency is not yet implemented in Iceberg must may come in the future support table where conflict are likely to appear
      - The concurrency control is handled by the catalog
    - - Updates on how the table is partitioned are possible without having to rewrite the entire table. Only the metadata in impacted.
    - - When you create a table, you can configure how to partition it using expressions, like day(event_ts). These expressions tell Iceberg how to derive partition values at write time.
      - The available partition transforms
        
        year, month, day, and hour – Partitions date/time values by hour, day, and so on
        
        bucket(N, any_col) – Pseudo-randomly divides rows across N buckets
        
        truncate(L, string_col) – Truncates a string value to length L
        
        truncate(W, numeric_col) – Truncates numbers to width W; for example truncate(100, 314) produces 300
    - - Copy-On-Write (COW)
        
        Best for tables with frequent reads, infrequent writes/updates, or large batch updates
        
        When a change is made to delete or update a particular row or rows, the datafiles with those rows are duplicated, but the new version has the updated rows
        
        This makes writes slower depending on how many data files must be re-written which can lead to concurrent writes having conflicts and potentially exceeding the number of reattempts and failing
        
        Pros
        
        :check: Fatest reads
        
        :check: Good for infrequent updates/deletes
        
        Cons
        
        :red_cross: Expensive writes
        
        :red_cross: Bad for frequent updates / deletes
      - Merge-On-Read (MOR)
        
        The file is not rewritten, instead the changes are written to a new file. Then when the data is read, the changes are applied or merged to the original data file to form the new state of the data during processing
        
        This makes writing the changes much quicker, but also means more work must be done when the data is read.
        
        If you delete a row, it gets added to a delete file and reconciled on each subsequent read till the files undergo compaction which will rewrite all the data into new files that won’t require the need for the delete file.
        
        If you update a row, that row is tracked via a delete file so future reads ignore it from the old data file and the updated row is added to a new data file.
        
        When using merge-on-read, it is recommended to have regular compaction jobs to impact reads as little as possible while still maintaining the faster write speeds.
    - - Apache Iceberg provides immutable snapshots, so the information for the table’s historical state is accessible, allowing you to run queries on the state of the table at a given point in time in the past
    - - Iceberg’s snapshot isolation allow you to query the data as it is, but it also reverts the table’s current state to any of those previous snapshots. Therefore, undoing mistakes is as easy as rolling back
    - - Handle schema evolution (adding/removing) a column, change the column type, etc ...
- - - - Obviously where the data are stored
      - Correspond to the leaves of the Iceberg tree structure
      - Store on well known storage -> HDFS or Object Storages
      - Data file format can be Parquet, ORC, Avro
        
        Parquet still the most common used as it is columnar format designed for heavy OLAP workload
    - - Delete files track which records in the dataset have been deleted.
      - Note
        that delete files are only supported in the Iceberg v2 format
      - Positional delete files
        
        Describe which rows of the dataset has been logically deleted
        
        Use at consumation time to not retrieve those rows
        
        Use the index in the data to locate the deleted row
      - Equality delete files
        
        Describe which rows of the dataset has been logically deleted
        
        Use at consumation time to not retrieve those rows
        
        Use the equality with some other value of the row to identify the deletion (primary key principle)
        
        A bit more flexible as it can reference multiple rows
        
        Note that in contrast to positional delete
        files, there is no reference to where these rows are located within the table
  - - - Manifest files keep track of files in the data layer (data files, delete files)
      - For each data files, manifest files also track some stats details (min, max per column)
      - 1 manifest file contains information on a subset datafiles
        
        Partition membership
        
        Records count
        
        Lower bound
        
        Upper bound
        
        Even if those information are also stored at the file level depending on the format choose, here the manifest file centralized the information for multiples files
        
        Those information are generated at the writing operation time by the engine
        
        In hive : stats are calculated by the metastore service and it's a pretty heavy operation, so stats are often/always outdated and so can't be used to optimize the query plan
        
        example : https://github.com/developer-advocacy-dremio/definitive-guide-to-apache-iceberg/blob/main/Resources/Chapter_2/manifest-file.json
    - - A manifest list is a snapshot of an Iceberg table at a given point in time
      - For the table at that point in time, it contains a list of all the manifest files, including the location, the partitions it belongs to, and the upper and lower bounds for partition columns for the datafiles it tracks
      - A manifest list contains an array of structs, with each struct keeping track of a single manifest file.
      - Example: https://github.com/developer-advocacy-dremio/definitive-guide-to-apache-iceberg/blob/main/Resources/Chapter_2/manifest-list.json
    - - A puffin file stores statistics and indexes about the data in the table that improve the performance of an even broader range of queries
      - Can store lot of statistic in the future but as of now, only the Theta sketch structure is supported bringing an approx number of distinct value per column
    - - Manifest lists are tracked by metadata files
      - Includes information about the table’s schema, partition information, snapshots, and which snapshot is the current one
      - Each time a change is made to an Iceberg table, a new metadata file is created and is registered as the latest version of the metadata file atomically via the catalog
      - Example: https://github.com/developer-advocacy-dremio/definitive-guide-to-apache-iceberg/blob/main/Resources/Chapter_2/metadata-file.json
  - - - Amazon s3 / ADLS object called version-hint.text. This way is called Hadoop catalog
      - Using Hive metastore, the table location property is used to store the location to the current metadata file
      - Using Nessie: same as for hive metastore but the property is called metadataLocation
      - AWS Glue
- - - - How the various components of Apache Iceberg work together when a READ query is executed
      - SELECT *
        FROM orders
        WHERE order_ts BETWEEN '2023-01-01' AND '2023-01-31'
      - Send the query to the engine
        
        1.a. The query engine requests the catalog for the current metadata file path.
        
        1.b. The engine opens and reads the metadata file
        
        The table schema (preparing internal memory structure for reading the data)
        
        The table partitioning scheme to know how the data are organized
        
        The current-snapshot-id in order to locate the accurate manifest list file path
        
        Relevant manifest list files
        
        1.c. Get Information from the manifest list file
        
        The manifest file path for the snapshot.
        
        Partition specific statistic : lower/upper bound of the partition columns
        
        Total number of datafiles/deletefiles and corresponding information at row level.
        
        1.d. Get information from the manifest file
        
        The query engine scans every entries, one per datafile tracked by this manifest file
        
        It compares the partition values for each datafiles with the query filters
        
        When the filter values match the partition value, the engine will check for all the records in this partition
        
        Retrieve relevant datafile paths (is the datafile relevant ? -> it is calculated based on lower - upper bound and null value count for each columns of the datafile)
        
        1.e. Get the data from the datafile
        
        Results:
        order_id customer_id order_amount order_ts
        1 125 321 20.50 2023-01-27 10:30:05 +00:00
    - - Apache Iceberg provides two ways to run time-travel queries: using a timestamp and using a snapshot ID
        
        Timestamp version:
        SELECT *
        FROM orders
        TIMESTAMP AS OF '2023-03-07 20:45:08.914'
        
        Snapshot version:
        SELECT *
        FROM orders
        VERSION AS OF 8333017788700497002
      - The first thing we need to understand is the history of the Iceberg table
        
        To analyze our order table’s history, we will query the history metadata table
        
        SELECT * FROM catalog.db.orders.history;
        
        | made_current_at | snapshot_id | parent_id | is_current_ancestor |
        | ---------------------------------- | ------------------------------- | ------ | ----- |
        | 2023-03-06 21:28:35.360 | 7327164675870333694 | null | true |
        | 2023-03-07 20:45:08.914 | 8333017788700497002 | 7327164675870333694 | true |
        | 2023-03-09 19:45:08.914 | 5139476312242609518 | 8333017788700497002 | true |
      - Send the query to the engine
        
        1.a. The query engine requests the catalog for the current metadata file path.
        
        1.b. The engine opens and reads the metadata file
        
        From the available list of snapshots, the engine will determine the particular snapshot specified in the time-travel query based on either the timestamp value or the snapshot ID.
        
        The table partitioning scheme to know how the data are organized
        
        The table schema (preparing internal memory structure for reading the data)
        
        Relevant manifest list files
        
        1.c. Get Information from the manifest list file
        
        The manifest file path for each snapshot within the manifest list.
        
        Partition specific statistic : lower/upper bound of the partition columns
        
        Total number of datafiles/deletefiles and corresponding information at row level.
        
        1.c. Get Information from the manifest list file
        
        The manifest file path for each snapshot within the manifest list.
        
        Partition specific statistic : lower/upper bound of the partition columns
        
        Total number of datafiles/deletefiles and corresponding information at row level.
        
        1.e. Get the data from the datafile
        
        Results:
        order id customer_id order_amount order_ts
        123 456 36.17 2023-03-07 08:10:23 +00:00
        
        1.d. Get information from the manifest file
        
        The query engine scans every entries, one per datafile tracked by this manifest file
        
        It compares the partition values for each datafiles with the query filters
        
        When the filter values match the partition value, the engine will check for all the records in this partition
        
        Retrieve relevant datafile paths (is the datafile relevant ? -> it is calculated based on lower - upper bound and null value count for each columns of the datafile)
  - - - -- Spark SQL
        CREATE TABLE orders (
        order_id BIGINT,
        customer_id BIGINT,
        order_amount DECIMAL(10, 2),
        order_ts TIMESTAMP
        )
        USING iceberg
        PARTITIONED BY (HOUR(order_ts))
      - First, the query is sent to the query engine for parsing. Then, since it is a CREATE statement, the engine will start creating and defining the table.
      - the engine starts creating a metadata file named v1.metadata.json in the data lake filesystem to store information about the table
      - Based on the information on the table path, /path/to/warehouse/db1/table1, the engine writes the metadata file
        
        Add the schema info + add an unique identifier for the table table-uuid
    - - We will input the following values into the table: an order_id of 123, a customer_id of 456, an order_amount of 36.17, and an order_ts of 2023-03-07 08:10:23.
      - INSERT INTO orders VALUES (
        123,
        456,
        36.17,
        '2023-03-07 08:10:23'
        )
      - First, the query is sent to the query engine for parsing. Then, Since this is an INSERT statement, the engine needs information about the table, such as its schema, to start with query planning.
      - It starts writing the new datafiles and the related metadata files
        
        The engine first writes the records as a Parquet datafile (Parquet is the default data file format) based on our hourly partitioning scheme defined in the metadata at the creation step
        
        If a sort order is defined then data are sorted before been written in the datafile.
        
        After writing the datafile, the engine creates a manifest file which contains statistical information, such as the upper and lower bounds of a column and the null value counts, in the manifest file
        
        Fun fact: the manifest file is written in Avro in the metadata folder
      - The engine creates a manifest list to keep track of the manifest file. If existing manifest files are associated with this snapshot, those will also be added to this new manifest list.
      - The engine creates a new metadata file, v2.metadata.json, with a new snapshot, s1, by considering the existing metadata file
      - The engine goes to the catalog again to ensure that no other snapshots were committed while this INSERT operation was being run. By doing this validation, Iceberg guarantees no interference in operations in a scenario where multiple writers write data concurrently
      - Upon completing the write, the engine commits atomically by switching the table’s metadata file pointer from the existing base version to the new version, v2.metadata.json, which now becomes the current metadata file.
    - - Such queries are usually run when you want to update an existing row if a specific value exists in the table, and if not, you just insert the new row
      - Send the query to the engine
        
        The query is first parsed by the query engine. In this case, since two tables are involved (stage and destination), the engine needs the data for both tables to start with the query planning.
      - Check the catalog
        
        The query engine first makes a request to the catalog to determine the current metadata file location and then reads it
      - Write datafiles and metadata files
        
        The query engine will read and load data in memory from both tables to determine the matching records :red_flag: It is using the READ process defined in the next section
        
        The engines is writing the merged records in a new datafile (due to the COW strategy)
        
        The engine creates a new manifest file that holds a reference to the filepath of these two datafiles. Additionally, various statistics about these datafiles, such as the lower and upper bounds of a column and the value counts, are included in the manifest file
        
        The engine then generates a new manifest list that points to the manifest file created in the previous step. It also tracks any existing manifest files and writes the manifest list to the data lake
        
        After that, the engine goes on to create a new metadata file, v3.metadata.json, with a new snapshot, s2, based on the previously current metadata file, v2.metadata.json, and the snapshots included as part of that, s0 and s1
      - Update the catalog
        
        The engine runs a check at this point to ensure that there are no write conflicts and then updates the catalog with the value of the latest metadata file, which is v3.metadata.json
    - - The catalog is then consulted to ensure consistency and integrity in the data and to write the data as per the defined partition strategies
      - The datafiles and metadata files are then written based on the query
      - the catalog file is updated to reflect the latest metadata, enabling subsequent read operations to access the most up-to-date version of the data
- - - - Doesn't required external service / process
      - Hadoop catalog works with HDFS or popular object storage (S3, ADLS...)
      - A version-hint.txt file is created only containing the name of the current snapshot ex: v[n].metadata.json
      - pros
        
        External system is not required
      - cons
        
        Not recommended for production usage
        
        Required that the filesystem rename operation to be atomic (HDFS, ADLS ok but S3 nope)
        
        As it can only use one warehouse directory (for listing tables). It cannot be split over several bucket or container resulting on scaling issue.
        
        Listing operation is a very costly operation on object storage
        
        No data/metadata separation which prevent from dropping the table without losing the data as well
        
        This isn’t possible if you use the Hadoop catalog, because even if the version-hint.txt file is deleted, the table still exists in the catalog definition since there are files in warehouse_dir/namespace1/table1/metadata. The current metadata file is still retrievable by doing a listing of the metadata directory and choosing the most recently created metadata file (going off the timestamp listed for the file in the filesystem).
    - - Keep a mapping between tables and the current metadata file using the table location.
      - pros
        
        Hive metastore is compatible with a wide range of tools
        
        Already deployed widely as historical used as Hive metastore for the hive tables
        
        Cloud agnostic
      - cons
        
        Needs to run an additional service on the side
        
        No multitable transaction (no consistency)
    - - Keep a mapping between tables and the current metadata file using the table property called metadata_location.
      - pros
        
        It's a managed service
      - cons
        
        Like the hive metastore, it doesn't support multitable transactions
        
        Not cloud agnostic, it could be a limitation in multi-cloud implementation
    - - Keep a mapping between tables and the current metadata file using the table property called metadataLocation.
      - pros
        
        Nessie comes with a git-like enabling the data-as-code concept. Meaning that data and its metadata can be versioned like source code.
        
        Multitable/multistatement transaction exits. So coherence can be guarantee.
        
        Cloud agnostic
      - cons
        
        Not widely support yet. Supported by Spark, Flink, Dreamio, Presto, Trino and PyIceberg
        
        Have to run a separate service as for the hive metastore
    - - The JDBC catalog leverages Java Database Connectivity (JDBC)–compliant data stores for Iceberg’s catalog interface
      - The JDBC catalog maps a table’s path to its current metadata file via a table property called metadata_location in the JDBC-compliant database,
      - prons
        
        Can be used with MySQL, PostgresSQL and those services are already widely deployed.
        
        cloud agnostic
        
        can be managed
      - cons
        
        Tools have to know how to talk in JDBC
  - - - It enables bulk migration of Apache Iceberg tables from one catalog to another without the need to copy data
      - The tool supports all commonly used catalogs in Apache Iceberg, such as AWS Glue, Nessie, Dremio Arctic, Hadoop, Hive, REST, JDBC, and any custom catalogs
      - 2 operations available
        
        migrate
        
        Facilitates the bulk migration. Tables are effectively migrated at the end -> no more tables exists in the source catalog.
        
        register
        
        Allow you to copy tables from a source catalog to a destination catalog. It doesn't remove the originals tables form the source.
        
        Need to ensure that there is no write at the same table on multiple catalog which can lead to miss updates / data losses or corruption.
      - It is recommended to not migrate while writings occurs. The tool capture a specific state of the table and use it for registering in the destination catalog.
      - usage exemple
        
        java -jar iceberg-catalog-migrator-cli-0.2.0.jar migrate
        --source-catalog-type GLUE
        --source-catalog-properties warehouse=s3a://bucket/gluecatalog/,ioimpl= org.apache.iceberg.aws.s3.S3FileIO
        --source-catalog-hadoop-conf fs.s3a.secret.key=$AWS_SECRET_ACCESS_KEY,fs.s3a.access.key=$AWS_ACCESS_KEY_ID : :
        --target-catalog-type NESSIE
        --target-catalog-properties uri=http://localhost:19120/api/v1,ref=main,warehouse= s3a://bucket/nessie/,io-impl=org.apache.iceberg.aws.s3.S3FileIO
        --identifiers db1.nominees
    - - It is possible to run a sparkSession with multiple catalogs spark.sql.catalog.NAME.xxx
      - register_table(target_table_name, current_metadata_file)
      - CALL target_catalog.system.register_table('target_catalog.db1.table1', '/path/to/source_catalog_warehouse/db1/table1/ metadata/xxx.json' )
- - - - We can query the table catalog.table.metadata_log_entries to get the snapshot prior to the disaster timestamp
      - Then we can restore the table as it was at given snapshot with the rollback command
      - SELECT snapshot_id
        FROM catalog.table.metadata_log_entries
        WHERE made_current_at < '2023-07-11 00:00:00'
        ORDER BY made_current_at ASC
    - - The metadata_log_entries metadata table keeps track of the evolution of the table by logging the metadata files generated during table updates
      - Fields
        
        timestamp = the exact date and time when the metadata was created.
        
        file = indicates the location of the datafile
        
        latest_snapshot_id = the identifier of the most recent snapshot at the time of metadata update.
        
        latest_schema_id = contains the ID of the schema being used when the metadata log entry was created. This gives context about the structure of the data at the time of the metadata update.
        
        latest_sequence_number = an incrementing count that helps track the sequence of metadata changes over the time.
    - - Understand the pattern of data additions to the table
      - SQL query that shows the count of each operation type over time
        
        SELECT
        operation,
        COUNT(*) AS operation_count,
        DATE(committed_at) AS date
        FROM
        catalog.table.snapshots
        GROUP BY
        operation,
        DATE(committed_at)
        ORDER BY
        date;
    - - Store detailed information about datafiles within table (location, format, partitioning strategy)
      - content = {0, 1, 2} where 0 = datafile, 1 = position delete file, 2 = equality delete file
      - file_path
      - file_format
      - spec_id = correspond to the partition spec id that the file adheres to
      - records_count = reports the number of records contained within each datafile
      - file_size_in_bytes
      - values_count
      - null_value_counts
      - lower_bounds = minimum value in each column
      - upper_bounds = maximum value in each columns
      - possible use cases
        
        Determining whether a partition should be rewritten
        
        If a partition has many small files => it is a candidate for a compaction
        
        example
        
        SELECT
        partition,
        COUNT(*) AS num_files,
        AVG(file_size_in_bytes) AS avg_file_size
        FROM
        catalog.table.files
        GROUP BY
        partition
        ORDER BY
        num_files DESC,
        avg_file_size ASC
        
        Identify which partitions need to be repair
        
        Calculate total size of a given snapshot
        
        Getting the list of files for a previous snapshot
    - - The manifests metadata table details each of the table’s current manifest files. This table offers an array of useful information that assists in understanding the table’s structure and changes over time
      - path = the filepath where the manifest is stored
      - length = the manifest file size
      - ....
      - possible use cases
        
        finding manifests that need rewriting
        
        summing the total number of files added per
        snapshot
        
        SELECT
        added_snapshot_id,
        SUM(added_data_files_count) AS total_added_data_files
        FROM
        catalog.table.manifests
        GROUP BY
        added_snapshot_id;
        
        finding snapshots where files were deleted
        
        determining whether the table is sorted well
    - - The partitions metadata table in Apache Iceberg provides a snapshot of how the data in a table is divided into distinct, nonoverlapping regions, known as partitions. Each row represents a specific partition within the table.
      - fields
        
        partition = the actual partition values, usually based on certain columns of your data.
        
        record_count = the total number of records within a given partition
        
        file_count = the total number of datafiles present in the partition (having too many small files impact the performance)
        
        spec_id the partition id
      - use cases
        
        finding how many files are in a partition
        
        summing the total size in bytes of a partition
        
        SELECT partition, SUM(file_size_in_bytes) AS partition_size FROM cata log.table.files GROUP BY partition
        
        finding the number of partitions per partition scheme
        
        you may want to see how many files are in a partition, because if a particular partition has a large number of files, it may be a candidate for compaction
        
        SELECT partition, file_count FROM catalog.table.partitions
    - - The all_data_files metadata table in Apache Iceberg provides comprehensive details about every datafile across all valid snapshots in the table.
      - fields
        
        file_path
        
        file_format
        
        spec_id
        
        partition
        
        record_count
        
        file_size_in_bytes
        
        coloumn_size = the size of each columns in byte
        
        value_counts / null_value_counts / nan_value_counts
        
        lower_bounds
        
        upper_bounds
        
        split_offset = provides info about split points within the file
        
        equality_ids = relates to equality deletes, help to identify rows deleted by an equality deletion
        
        sort_order_id = the id of the sort order on the datafile
        
        readable_metrics = derived field that provides human-readable file metadata (column size, value counts, null counts, lower / upper bounds)
      - use cases
        
        finding the largest table across all snapshots
        
        finding the total file size across all snapshots
        
        assessing partitions across snapshots
    - - fields
        
        content = {0, 1} 0 = tracking a manifest file ; 1 = tracking delete file
        
        path_field = path to the manifest file
        
        length = size of the manifest file in bytes
        
        partitions_spec_id = the id of the partition specification
        
        added_snapshot_id = the ID of the snapshot when created
        
        added_data_files_count
        
        partitions_summaries = an array of structures where each one provides a summary for a specific partition
        
        reference_snapshot_id = the ID of the snapshot that this record is associated with. You’ll see a manifest listed once for each snapshot it was valid for
      - use case
        
        finding all manifests for a particular snapshot
        
        monitoring the growth of manifests from
        snapshot to snapshot
        
        getting the total size of all valid manifests
      - The all_manifests metadata table in Apache Iceberg provides detailed insights into every manifest file across all valid snapshots in the table.
  - - - Isolation changes at the table level
        
        Create git-branch for specific tables. Each branch contains a full history of changes made to that table.
      - Isolation changes at the catalog level
        
        manage the whole datalake as a single big entity, capturing changes acrosss multiples table
      - advantages / drawbacks
        
        Table-level isolation provides granular control and flexibility for individual tables but might become complex to manage in a large-scale data environment
        
        Catalog-level isolation provides a comprehensive,
        unified view of changes but might be overkill for small-scale or single-table scenarios.
- - - - --packages
      - --jars
    - - spark.sql.catalog.my_catalog=org.apache.iceberg.spark.SparkCatalog
        
        with Hive implementation
        
        spark.sql.catalog.hive_catalog = org.apache.iceberg.spark.SparkCatalog
        
        spark.sql.catalog.hive_catalog.type = hive
        
        spark.sql.catalog.hive_catalog.uri = thrift://metastore-host:port)
        
        with Hadoop implementation
        
        spark.sql.catalog.hive_catalog = org.apache.iceberg.spark.SparkCatalog
        
        spark.sql.catalog.hive_catalog.type = Hadoop
        
        spark.sql.catalog.hive_catalog.warehouse = /path/to/warehouse
      - spark.sql.catalog.hive_spark_catalog = org.apache.iceberg.spark.SparkSessionCatalog
        
        This could benefit scenarios when you want to use Iceberg tables seamlessly alongside non-Iceberg tables in your Spark session