GCS Files to Iceberg Lakehouse Pipeline
A highly configurable, orchestrator-agnostic Spark ETL pipeline built to ingest raw logs, files, and partitions from Google Cloud Storage directly into an Apache Iceberg-based analytical data lakehouse.
Deployment Orchestration: This pipeline executes on Dataproc Serverless or GKE and is scheduler-agnostic. Workflows can be triggered hourly, daily, or dynamically using Apache Airflow, Prefect, or Dagster by supplying table-level configuration profiles in YAML or HOCON.
Data Pipeline Execution Workflow
Modular Pipeline Architecture
The pipeline leverages a composition DSL (~>) to decouple stages completely. FilesExtract scans directories, handles custom checkpoint comparison, and isolates partitions. CommonTransformations manages schema standardization, mapping, and filters. MultiLoad loads data into the target Iceberg layout and publishes ingestion metrics.
1
2val etl: ETL = FilesExtract ~> CommonTransformations ~> MultiLoad(IcebergLoad, BatchLoadMetricsPublisher)
3Ingestion Types
The ingestion strategy is fully parameter-driven, allowing you to choose how incoming GCS partitions are read and processed:
| Ingestion Mode | Description | Best Use Case |
|---|---|---|
full | Reads all glob-matched files on every run. No checkpoint is written. Combined with load.save-mode = Overwrite, this fully refreshes the target table. | Small lookup dimensions, master datasets, or full historical backfills. |
last | Identifies partitions dynamically using regex and reads only files matching the lexicographically or chronologically largest partition. | Latest snapshot landing zones where only the most recent daily output is required. |
incremental | Checkpoint-driven path detection. Compares partition values in file paths to only read files and directories newer than the last successful execution. | High-frequency data streams, continuous raw event logging, or append-only transactional tables. |
Key Ingestion Controls
Preventing Partial Data Ingestion
success-file-filter = trueGuards downstream systems by checking for upstream completeness. If enabled, the pipeline only processes files in directories containing a _SUCCESS marker. This prevents the ingestion of partial writes caused by upstream failures or mid-execution writes.
Mitigating Runaway Executions
max-batch-size = [count]A critical memory and cost control feature. During massive catchups or historical re-processing, loading all files in a single execution might overwhelm Spark executors. max-batch-size restricts the number of file paths processed per execution, preventing runaway execution costs and avoiding Out-Of-Memory (OOM) errors.
How Custom Files2Iceberg Outperforms Native GCP Services
Standard cloud tools often ingest entire folders or force you to run separate Spark scripts for data merging and deduplication. Our custom Files2Iceberg pipeline is designed specifically for efficient, governed lakehouse creation:
- Smart Skip & Scan Reduction: Native services often scan all objects in a bucket to identify changes.
Files2Icebergcompares file path metadata against localized custom checkpoints, bypassing file scanning entirely to reduce API call overhead and storage lookup costs. - Built-In Deduplication: Instead of writing raw outputs and scheduling secondary consolidation scripts, you can specify primary keys in configuration to execute direct
MERGE INTOupdates with precombine priority conflict resolution. - Custom Schema Stability: Map missing columns with NULL values dynamically. This protects target tables from schema breaks if upstream file extractors omit optional attributes.
Pipeline Configurable Parameters
Control the pipeline's behavior end-to-end directly using configuration fields. No Scala changes or recompilations are needed.
1. Extraction Parameters
HCS/GCS glob patterns referencing source logs or directories. Supports standard wildcards like *, ?, or {a,b}.
Source format. Defaults to parquet. Also supports orc, avro, json, csv, and delta.
When set to true, recursively traverses all subdirectories matching the path glob pattern.
Flat map of generic key-value parameters passed directly to the Spark DataFrameReader (e.g. delimiters or csv headers).
2. Partition & Selection Strategy
Inclusion strategy. Must be set to either full, last, or incremental.
Method to evaluate folder paths. Supports string (lexicographical check) or date (chronological parsing).
Regex containing a single capture group to isolate the partition token from the GCS file URI path.
Format (e.g. yyyy-MM-dd) required to parse and sort partition tokens chronologically.
When true, aborts the run if any scanned file fails the partition regex check. If false, skips them.
3. Checkpoint & Control
Configures the maximum number of historical checkpoints to retain in the metadata path before pruning.
Toggles checkpoint writing. Set to false to run manual backfills without affecting production bookmarks.
Soft threshold on total processing paths. Limits file load count per execution batch.
4. Load Parameters
The target fully qualified Iceberg catalog coordinate (e.g., prod_catalog.db.table_name).
Save behavior. Set to Append (common for incremental) or Overwrite.
Defines primary key columns to execute MERGE INTO upserts rather than double-insert appends.
List of fields utilized to partition the target physical Iceberg files on GCS storage.
HOCON Config Profile Example
Below is a configuration file showing how easy it is to specify file source paths, partition regex, transformation, and target Iceberg properties:
1# Incremental GCS Files ingestion using HOCON configuration profile
2extract {
3 path = "gs://my-lakehouse-bucket/raw/clicks/dt=*/*.parquet"
4 format = "parquet"
5 recursive-file-lookup = false
6
7 batch-file {
8 ingestion-type = "incremental" # Can be: incremental, full, last
9 partition-type = "date" # Can be: date, string
10 pattern = ".+/dt=(\\d{4}-\\d{2}-\\d{2})/.+"
11 date-format = "yyyy-MM-dd"
12 strict-pattern = true
13 success-file-filter = true # Only ingest directory if _SUCCESS file exists
14 }
15
16 keep-checkpoints = 50
17 save-checkpoint = true
18 max-batch-size = 100 # Caps processing paths per run to avoid OOM
19}
20
21transform {
22 filter = ["click_id IS NOT NULL", "user_id != ''"]
23 rename-columns {
24 temp_id = "temporary_id"
25 }
26 cast.columns {
27 user_id = "BIGINT"
28 }
29}
30
31load {
32 iceberg.table = "prod_catalog.raw.clickstream"
33 save-mode = "Append"
34 primary-key = ["click_id"]
35 partition-by = ["dt"]
36 iceberg.precombine-field = "click_time"
37}
38
39metadata {
40 path = "gs://my-lakehouse-bucket/metadata/clickstream/"
41}Architectural & Cost Comparisons
Compare the performance and cost benefits of our custom Spark pipeline against native GCP ingestion alternatives:
