Big Data Fundamentals: data lake example

Building a Production-Grade Data Lake with Apache Iceberg: A Deep Dive

Introduction

The increasing demand for real-time analytics and machine learning on massive datasets presents a significant engineering challenge: managing data at scale while maintaining data quality, consistency, and query performance. We recently faced this at scale while building a fraud detection system for a large e-commerce platform. The system needed to ingest and analyze billions of events daily, encompassing user behavior, transaction details, and external risk scores. Traditional Hive-based data lakes struggled with schema evolution, concurrent writes, and query latency, especially during peak hours. This led us to adopt Apache Iceberg as the foundation for our data lake, and this post details our architecture, performance optimizations, and operational learnings. We operate in a multi-tenant AWS environment, leveraging EMR, S3, and a mix of Spark and Flink for processing. Data volume is approximately 50TB/day ingested, with a total lake size exceeding 500TB. Query latency requirements range from seconds for interactive dashboards to minutes for batch reporting.

What is Apache Iceberg in Big Data Systems?

Apache Iceberg is an open table format for huge analytic datasets. Unlike Hive tables which rely on the metastore for metadata, Iceberg manages its own metadata in a hierarchical structure, providing ACID transactions, schema evolution, and time travel capabilities. From an architectural perspective, it decouples metadata management from the storage layer (S3, GCS, ADLS). Iceberg tables are defined by a manifest list, which points to manifest files, which in turn point to data files. This layered approach allows for efficient metadata operations and snapshot isolation.

Protocol-level behavior is crucial. Iceberg uses a snapshot isolation model. Readers always see a consistent view of the table, even during concurrent writes. Writes create new snapshots, and the manifest list is atomically updated to point to the new snapshot. This ensures that queries always operate on a consistent state of the data. We primarily use Parquet as the underlying data format due to its efficient compression and columnar storage, but Iceberg supports ORC and Avro as well.

Real-World Use Cases

1. Change Data Capture (CDC) Ingestion: We ingest data from multiple transactional databases using Debezium and Kafka. Iceberg’s transactional capabilities ensure that updates and deletes are applied atomically, preventing data corruption and ensuring consistency.
2. Streaming ETL: Flink continuously processes streaming data from Kafka, performing transformations and writing the results to Iceberg tables. The ability to append data efficiently and handle schema evolution is critical for this use case.
3. Large-Scale Joins: Our fraud detection system requires joining multiple large tables (user profiles, transaction history, risk scores). Iceberg’s metadata management and partitioning strategies significantly improve query performance.
4. Schema Validation & Evolution: As business requirements change, we frequently need to add or modify columns in our tables. Iceberg’s schema evolution capabilities allow us to do this without disrupting existing queries.
5. ML Feature Pipelines: We use Iceberg tables as the source for our machine learning feature pipelines. The time travel feature allows us to recreate historical feature sets for model retraining and backtesting.

System Design & Architecture

graph LR
    A[Debezium/Kafka] --> B(Flink Streaming Job);
    C[Kafka] --> D(Spark Batch Job);
    B --> E[Iceberg Table (S3)];
    D --> E;
    E --> F{Presto/Trino};
    E --> G[Spark SQL];
    F --> H[Dashboards/Reporting];
    G --> I[ML Feature Store];
    subgraph Data Lake
        E
    end
    subgraph Ingestion
        A
        C
        B
        D
    end
    subgraph Consumption
        F
        G
        H
        I
    end

This diagram illustrates our end-to-end pipeline. Data is ingested via CDC (Debezium/Kafka) and batch loads (Kafka/Spark). Both streams converge into Iceberg tables stored in S3. Presto/Trino and Spark SQL are used for querying the data.

For our EMR setup, we utilize a dedicated metastore cluster running MySQL for Iceberg metadata. We partition our tables by event time (year, month, day) to optimize query performance. We also leverage S3 lifecycle policies to tier older data to Glacier for cost savings.

Performance Tuning & Resource Management

Performance tuning is critical for maintaining acceptable query latency. Here are some key strategies:

• File Size Compaction: Small files can significantly degrade query performance. We schedule regular compaction jobs to combine small files into larger, more efficient files. We aim for file sizes between 128MB and 256MB.
• Partitioning: Choosing the right partitioning strategy is crucial. We found that partitioning by event time provides good performance for most of our queries.
• Data Skipping: Iceberg’s metadata allows for efficient data skipping. We leverage this by using appropriate filters in our queries.
• Spark Configuration:
  • spark.sql.shuffle.partitions: Set to 200 to balance parallelism and overhead.
  • fs.s3a.connection.maximum: Set to 1000 to maximize S3 throughput.
  • spark.driver.memory: Adjusted based on the size of the query and the amount of data being processed. Typically 16-32GB.
  • spark.executor.memory: Adjusted based on the size of the query and the amount of data being processed. Typically 64-128GB.
• Presto/Trino Configuration:
  • query.max-memory: Adjusted based on the complexity of the query.
  • query.max-concurrent-queries: Controlled to prevent resource exhaustion.

Failure Modes & Debugging

Common failure modes include:

• Data Skew: Uneven data distribution can lead to performance bottlenecks. We use techniques like salting to mitigate data skew.
• Out-of-Memory Errors: Large joins or aggregations can exhaust memory resources. We increase memory allocation or optimize the query to reduce memory usage.
• Job Retries: Transient errors (e.g., network issues) can cause jobs to fail. We configure automatic retries with exponential backoff.
• DAG Crashes (Flink): Complex Flink jobs can sometimes crash due to bugs or resource constraints. We use Flink’s savepoints to recover from failures.

Debugging tools:

• Spark UI: Provides detailed information about Spark jobs, including task execution times, memory usage, and shuffle statistics.
• Flink Dashboard: Provides real-time monitoring of Flink jobs, including throughput, latency, and error rates.
• Datadog: Used for monitoring system metrics (CPU usage, memory usage, disk I/O) and alerting on anomalies.
• Iceberg CLI: Useful for inspecting table metadata and verifying data consistency.

Data Governance & Schema Management

We use the Hive Metastore as a central metadata catalog, integrating it with Iceberg. We also leverage a schema registry (Confluent Schema Registry) to manage schema evolution. All schema changes are versioned and validated before being applied to the Iceberg tables. We enforce data quality checks using Great Expectations, validating data types, ranges, and completeness. Backward compatibility is maintained by using schema evolution features like adding optional columns.

Security and Access Control

We use AWS Lake Formation to manage access control to our data lake. Lake Formation allows us to define fine-grained access policies based on users, groups, and data attributes. Data is encrypted at rest using S3 encryption and in transit using TLS. We also enable audit logging to track data access and modifications.

Testing & CI/CD Integration

We use a combination of unit tests, integration tests, and end-to-end tests to validate our data pipelines. Great Expectations is used for data quality testing. DBT tests are used for data transformation testing. We use Apache Nifi unit tests to validate individual data flow components. Our CI/CD pipeline is automated using Jenkins, with automated regression tests running on every code commit.

Common Pitfalls & Operational Misconceptions

1. Ignoring File Size: Too many small files kill performance. Symptom: Slow query times. Mitigation: Implement regular compaction jobs.
2. Incorrect Partitioning: Poor partitioning leads to data skew and inefficient queries. Symptom: Uneven task execution times in Spark. Mitigation: Analyze query patterns and choose an appropriate partitioning strategy.
3. Lack of Schema Enforcement: Schema drift can lead to data quality issues. Symptom: Data validation failures. Mitigation: Implement schema validation using a schema registry.
4. Insufficient Resource Allocation: Under-provisioned resources can lead to performance bottlenecks and job failures. Symptom: Out-of-memory errors. Mitigation: Monitor resource usage and adjust allocation accordingly.
5. Overlooking Metadata Management: Ignoring Iceberg’s metadata features (data skipping, snapshot isolation) can lead to suboptimal performance. Symptom: Slow query times, data inconsistencies. Mitigation: Leverage Iceberg’s metadata features in your queries and data pipelines.

Enterprise Patterns & Best Practices

• Data Lakehouse vs. Warehouse: We’ve adopted a data lakehouse approach, combining the flexibility of a data lake with the reliability and performance of a data warehouse.
• Batch vs. Micro-Batch vs. Streaming: We use a combination of all three processing modes, depending on the specific use case.
• File Format Decisions: Parquet is our preferred format due to its efficiency, but we consider ORC for specific workloads.
• Storage Tiering: We use S3 lifecycle policies to tier older data to Glacier for cost savings.
• Workflow Orchestration: Airflow is used to orchestrate our data pipelines, providing scheduling, monitoring, and alerting capabilities.

Conclusion

Apache Iceberg has proven to be a valuable addition to our data lake architecture, addressing the challenges of schema evolution, concurrent writes, and query performance. By carefully tuning our system and implementing robust data governance practices, we have built a reliable and scalable data platform that supports our growing analytics and machine learning needs. Next steps include benchmarking new Iceberg configurations, introducing schema enforcement at the ingestion layer, and migrating to a more granular partitioning strategy.