# Integration with AWS Glue Catalog source: https://docs.chalk.ai/docs/iceberg-glue ## Scan Iceberg tables from an AWS Glue Catalog and write query results back to them. ### Introduction Apache Iceberg is a high-performance table format designed for managing large, evolving datasets, providing features such as schema evolution and time travel. The AWS Glue Catalog, on the other hand, is a fully managed metadata catalog that simplifies data discovery and schema management for data lakes. Chalk can both scan Iceberg tables registered in a Glue catalog, using chalkdf.DataFrame.scan_iceberg, and write query results back to Glue-cataloged Iceberg tables, using write_to destinations. ### Scanning Iceberg Data from Glue chalkdf.DataFrame.scan_iceberg builds a DataFrame backed by an Iceberg table that is registered in a catalog: ``` from chalkdf import DataFrame df = DataFrame.scan_iceberg( "banking.transactions", storage_options={ "catalog": "glue", "catalog_id": "123456789012", "region_name": "us-east-1", "warehouse": "s3://your-bucket/warehouse", }, ) ``` The method takes the following arguments: - table: The catalog-qualified identifier of the Iceberg table to read from. For Glue, this is formatted like database_name.table_name. - schema: Optional. A pyarrow.Schema for the table. When omitted, the schema is inferred from the Iceberg catalog. - storage_options: A mapping of catalog configuration properties (see below). When omitted, ambient catalog configuration from the engine's environment is used. - snapshot_id: Optional. Pins the scan to a specific Iceberg snapshot id. When omitted, the table's current snapshot at query planning time is used. The following storage_options keys configure a Glue-backed scan: | Key | Description | | ------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | | `catalog` | The catalog type. Use `"glue"` for the AWS Glue Catalog. | | `catalog_id` | The AWS account id that owns the Glue catalog. | | `region_name` | The AWS region of the Glue catalog. | | `warehouse` | The S3 URI of the Iceberg warehouse root (where the table data files live). | | `client.assume-role.arn` | Optional. An IAM role ARN to assume via STS for catalog and data access. When absent, ambient AWS credentials are used (instance profile / IRSA / environment variables). | ### Example Scanning is typically done inside a resolver marked with static=True, which indicates to the Chalk query planner that the resolver can be expanded at query planning time rather than query execution time. Use .rename to map the table's column names to your Chalk feature names: ``` import chalkdf from chalk import offline from chalk.features import DataFrame from src.features import Transaction @offline(static=True) def read_transactions() -> DataFrame[Transaction.id, Transaction.amount, Transaction.ts]: return ( chalkdf.DataFrame.scan_iceberg( "banking.transactions", storage_options={ "catalog": "glue", "catalog_id": "123456789012", "region_name": "us-east-1", "warehouse": "s3://your-bucket/warehouse", }, ) .select("id", "amount", "ts") .rename({ "id": "transaction.id", "amount": "transaction.amount", "ts": "transaction.ts", }) ) ``` Because scan_iceberg returns a plain chalkdf.DataFrame, you can also test scans locally — .run() executes the plan and returns a materialized result: ``` out = chalkdf.DataFrame.scan_iceberg( "banking.transactions", storage_options={...}, ).run() print(out.to_arrow()) ``` ### Pushdown Filters and Projections Chalk's query planner pushes projections and filters down into the Iceberg scan when possible. Columns that are not selected are never read from storage, and filters reduce the amount of data scanned. For correctly partitioned Iceberg tables, filter pushdown is applied automatically at runtime using the table's partition spec, and the full range of Iceberg partitioning transforms is supported. No additional configuration is required. For example, when scanning an Iceberg table with a filter like event_timestamp > '2024-06-01 10:35:00', if the table is partitioned by the column transform event_date = day(event_timestamp) then Iceberg will only scan the partitions where event_date >= '2024-06-01'. ``` from chalk.features import _ df = ( chalkdf.DataFrame.scan_iceberg("banking.transactions", storage_options={...}) .filter(_.amount > 100) .select("id", "amount") ) ``` ### Permissions To successfully query Iceberg data through AWS Glue, ensure that the IAM role or user used in your AWS credentials has the following permissions: ### Catalog Access Permissions These permissions allow access to the AWS Glue metadata: ``` glue:GetDatabase glue:GetTable glue:GetPartition glue:GetTableVersion glue:GetTableVersions ``` ### Underlying Data Access Permissions These permissions allow reading of the actual data stored in your data lake (e.g., in Amazon S3): ``` s3:GetObject s3:ListBucket ``` Properly configuring these permissions is crucial to ensure that your queries can access both the Glue catalog metadata and the underlying data without encountering authorization issues. ### Writing Query Results to Glue with write_to Offline queries and scheduled queries can append their outputs directly to an Iceberg table in a Glue catalog by passing an iceberg+... destination URI as the write_to argument: ``` from chalk.client import ChalkClient ChalkClient().offline_query( input={Transaction.id: txn_ids}, output=[Transaction.amount, Transaction.ts], write_to="iceberg+glue+s3://your-bucket/warehouse/banking/transaction_features?region=us-east-1&account=123456789012", ) ``` The same URI works as the write_to argument of a ScheduledQuery: ``` from chalk import ScheduledQuery ScheduledQuery( name="export-transaction-features", schedule="0 6 * * *", output=[Transaction.amount, Transaction.ts], write_to="iceberg+glue+s3://your-bucket/warehouse/banking/transaction_features?region=us-east-1&account=123456789012", ) ``` Writes append an Iceberg snapshot to the table. If the table does not exist yet, it is created in the Glue catalog on first write. Malformed URIs and unsupported options are rejected at query planning time, before any data is written. ### Destination URI Format Two URI shapes are supported: - Glue catalog: iceberg+glue+s3://{bucket}/{warehouse-path}/{database}/{table} — the final two path segments are the Glue database and table name; everything before them is the S3 warehouse root. - Direct table root (no catalog service): iceberg+file+s3://{bucket}/{path}/{table} or iceberg+local:///{abs-path}/{table} — the table's metadata is managed directly at the given storage location as versioned metadata.json files (see Writing to a Table Root below). ### Query Parameters | Parameter | Description | | ------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------- | | `region` | The AWS region of the Glue catalog and warehouse bucket. | | `account` | The AWS account id that owns the Glue catalog. | | `role_arn` | Optional. An IAM role ARN to assume via STS for catalog and storage access. | | `partition` | Optional. Comma-separated Spark-style partition transforms applied when the table is created (see below). | | `mode` | Optional. Only `append` (the default) is supported today. | | `aws_access_key_id` / `aws_secret_access_key` / `aws_session_token` | Optional. Explicit static AWS credentials, e.g. for writing to a Glue catalog in a different AWS account. | Credential parameter values are treated as secrets: they are redacted from any logged or displayed form of the query plan. ### Partitioning The partition parameter accepts a comma-separated list of transforms applied at table creation: - bucket(N, col) — hash-bucket col into N buckets - truncate(W, col) — truncate col to width W - year(col), month(col), day(col), hour(col) — temporal truncation of a timestamp column - a bare column name — identity partitioning For example: ``` iceberg+glue+s3://your-bucket/warehouse/banking/transaction_features?region=us-east-1&account=123456789012&partition=day(ts),bucket(16,id) ``` For an existing table, the requested spec is reconciled against the table's current partition spec, and the write is rejected if they are incompatible. ### Writing to a Table Root When the destination has no catalog service — the iceberg+file+... and iceberg+local://... URI forms — Chalk manages the table's metadata directly at the table root, using the standard Iceberg metadata layout: ``` {table-root}/ data/ # parquet data files metadata/ 00000-{uuid}.metadata.json # one metadata version per commit 00001-{uuid}.metadata.json snap-*.avro, *-m0.avro # snapshot and manifest files version-hint.text # the current metadata version number ``` The first write creates the table (writing metadata version 00000). Each subsequent write appends an Iceberg snapshot: the new data files are written under data/, the next NNNNN-{uuid}.metadata.json version is written, and version-hint.text is updated to point at it. Commits use an optimistic version check — if version-hint.text advanced while the write was in progress (a concurrent writer), the commit fails rather than clobbering the other write. Storage access uses the ambient credentials of the engine environment (e.g. IRSA / instance credentials, or AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY / AWS_REGION environment variables). Because this is the standard Iceberg static-table layout, the resulting table is readable by any Iceberg client that can read a metadata.json — for example PyIceberg's StaticTable, DuckDB's iceberg_scan, or Spark — as well as by Chalk itself: chalkdf.DataFrame.scan_iceberg accepts a direct table-root URI in place of a catalog-qualified name: ``` df = chalkdf.DataFrame.scan_iceberg("s3://your-bucket/exports/events") ``` ### Required IAM Permissions In addition to the read permissions above, writing requires: ``` glue:CreateTable glue:UpdateTable s3:PutObject ``` ### Adding Chalk Datasets to Glue To add Chalk datasets to Glue, you can use the Dataset.write_to method. ``` from chalk.integrations import GlueCatalog catalog = GlueCatalog( name="aws_glue_catalog", aws_region="us-west-2", catalog_id="123", aws_role_arn="arn:aws:iam::123456789012:role/YourCatalogueAccessRole", ) dataset.write_to(destination="database.table_name", catalog=catalog) ``` This will create a table referencing the dataset in specified location in the Glue catalog, making it available for querying using tools like AWS Athena. ### Required IAM Permissions To write to a Glue catalog, the IAM role or user used in your AWS credentials must have the following permissions: ``` glue:CreateTable ``` Note: this 'create table' operation runs from your client, not from the Chalk platform.