Constructing unified knowledge pipelines with Apache Iceberg and Apache Flink

You’ll be able to course of real-time knowledge out of your knowledge lake with Amazon Managed Service for Apache Flink with out sustaining two separate pipelines. But many groups do precisely that, and the price provides up quick. On this publish, you construct a unified pipeline utilizing Apache Iceberg and Amazon Managed Service for Apache Flink that replaces the dual-pipeline strategy. This walkthrough is for intermediate AWS customers who’re comfy with Amazon Easy Storage Service (Amazon S3) and AWS Glue Information Catalog however new to streaming from Apache Iceberg tables.

The twin-pipeline drawback

This dual-pipeline strategy creates three issues:

• Double the infrastructure prices. You run and pay for 2 separate compute environments, two storage layers, and two units of monitoring. For instance, if you happen to’re spending $10,000/month on separate streaming and batch infrastructure, a significant portion of that spend is pure duplication.
• Information synchronization points. Your batch and streaming shoppers learn from totally different copies of the info, processed at totally different occasions. When a transaction exhibits up in your real-time dashboard however not in your batch report (or vice versa), debugging the inconsistency takes hours.
• Operational complexity. Two pipelines imply two deployment processes, two failure modes to observe, and two units of schema evolution to handle. Your staff spends time reconciling programs as an alternative of constructing options.

The place this sample suits

Earlier than diving into the implementation, contemplate whether or not streaming out of your knowledge lake is the correct strategy in your use case.

Streaming from Apache Iceberg tables works nicely when you want knowledge out there inside seconds to minutes and also you question current knowledge regularly, a number of occasions per hour. Frequent situations embrace:

• Operational knowledge shops — Stream buyer profile updates to serve downstream functions like advice engines. When a buyer updates their preferences, these adjustments attain your operational knowledge retailer inside seconds.
• Fraud detection — Stream transactions for fast evaluation. Begin with a 3-second monitor interval and alter based mostly in your detection accuracy wants.
• Stay dashboards — Energy real-time analytics instantly out of your lake. That is the strongest start line if you happen to’re evaluating the strategy for the primary time, as a result of the suggestions loop is fast and simple to validate.
• Occasion-driven architectures — Set off downstream processes based mostly on knowledge adjustments in your Apache Iceberg tables.

Batch processing stays more cost effective when you course of knowledge as soon as per day or much less, otherwise you primarily question historic knowledge. Batch queries on Apache Iceberg tables value much less as a result of they don’t require a steady Apache Flink runtime.

How Apache Iceberg solves this

Apache Iceberg’s snapshot-based structure removes the necessity for a separate streaming pipeline. Consider snapshots like Git commits in your knowledge. Every time you write knowledge to your Iceberg desk, Iceberg creates a brand new snapshot that factors to the brand new knowledge recordsdata whereas preserving references to current recordsdata. Apache Flink reads solely the adjustments between snapshots (the brand new recordsdata that arrived after the final checkpoint), quite than scanning your entire desk. Atomicity, Consistency, Isolation, Sturdiness (ACID) transactions forestall your concurrent reads and writes from producing partial or inconsistent outcomes. For instance, in case your batch extract, remodel, and cargo (ETL) job is writing 10,000 data whereas your Flink software is studying, ACID transactions imply that your streaming question sees both the whole batch of 10,000 data or none of them, not a partial set that would skew your analytics.

The result’s a single pipeline that handles each real-time and batch entry from the identical knowledge, by way of the identical storage layer, with the identical schema.

Resolution structure

Your structure makes use of 4 AWS companies and one open supply desk format working collectively. The next diagram exhibits how these parts join, changing the dual-pipeline sample proven earlier with a single unified circulation.

Your supply knowledge lands in Amazon S3 as Apache Iceberg desk recordsdata. AWS Glue Information Catalog tracks the metadata and schema. When new knowledge arrives, Apache Iceberg creates a brand new snapshot that your software detects. Your Flink software displays these snapshots and processes new data incrementally, studying solely the recordsdata that arrived after the final checkpoint, not your entire desk.

You utilize 4 predominant parts:

• Amazon S3 — Foundational storage layer in your knowledge lake
• Information Catalog — Metadata and schema administration for Apache Iceberg tables
• Apache Iceberg — Desk format with snapshot-based streaming capabilities
• Amazon Managed Service for Apache Flink — Stream processing and incremental consumption

Essential notices

Earlier than implementing this resolution, consider these dangers in your surroundings:

• Information safety: Streaming from knowledge lakes exposes knowledge to further processing programs. Classify your knowledge earlier than implementation—buyer profile updates and transaction knowledge sometimes comprise personally identifiable info (PII) and deal with them as confidential. Apply encryption at relaxation and in transit for confidential knowledge. Key dangers embrace unauthorized knowledge entry by way of misconfigured Amazon S3 bucket insurance policies or overly permissive IAM roles. Mitigations: use the resource-scoped IAM coverage and TLS-enforcing bucket coverage supplied within the Safety part.
• Information integrity: Misconfigured checkpoints or schema adjustments throughout streaming can result in knowledge inconsistency. Mitigations: allow exactly-once processing semantics and check schema evolution in a non-production surroundings first.
• Compliance: Confirm that real-time knowledge processing meets your regulatory necessities. For workloads topic to HIPAA, affirm that you just use HIPAA Eligible Companies and have a Enterprise Affiliate Settlement (BAA) with AWS. For PCI-DSS or GDPR workloads, overview the related compliance documentation on the AWS Compliance web page. Implement knowledge retention insurance policies that comply together with your regulatory framework.
• Value: Practically steady streaming incurs ongoing compute prices. Monitor utilization to keep away from surprising fees. Value estimates on this publish are based mostly on pricing as of March 2026 and would possibly change. Confirm present pricing on the related AWS service pricing pages.
• Operational: Pipeline failures would possibly influence downstream programs. Implement monitoring and alerting earlier than working in manufacturing.

Conditions

Earlier than you start, just be sure you have the next in place. This walkthrough assumes intermediate Python abilities (comfy with features, error dealing with, and surroundings variables), primary Apache Flink ideas (streaming in comparison with batch processing), and primary AWS Identification and Entry Administration (AWS IAM) data (creating roles and attaching insurance policies). Plan for roughly 90–120 minutes, together with setup, implementation, and testing. First-time setup would possibly take longer as you obtain dependencies and configure AWS sources. Anticipated AWS prices: roughly $5–10 if you happen to full the walkthrough inside 2 hours and clear up sources instantly afterward. The first value driver is Amazon Managed Service for Apache Flink runtime ($0.11/hour per Kinesis Processing Unit (KPU)). You’ll be able to decrease prices by stopping your software when not in use.

• An AWS account with AWS IAM permissions for: s3:GetObject, s3:PutObject, s3:ListBucket in your knowledge bucket; glue:GetDatabase, glue:GetTable for catalog entry; and flink:CreateApplication, flink:StartApplication for Amazon Managed Service for Apache Flink
• An current Amazon S3 bucket in your knowledge lake
• An AWS Glue Information Catalog database configured
• Apache Flink 1.19.1 put in regionally
• Python 3.8 or later
• Java 11 or a newer model
• AWS Command Line Interface (AWS CLI) configured with credentials (aws configure)

Required Java Archive (JAR) dependencies

You want a number of JAR recordsdata as a result of your Flink software coordinates between totally different programs—Amazon S3 for storage, AWS Glue for metadata, Hadoop for file operations, and Apache Iceberg for the desk format. Every JAR handles a selected a part of this integration. Lacking even one causes ClassNotFoundException errors at runtime.

• iceberg-flink-runtime-1.19-1.6.1.jar — Core Apache Iceberg integration with Apache Flink
• iceberg-aws-bundle-1.6.1.jar — AWS-specific Apache Iceberg performance for Amazon S3 and AWS Glue
• flink-s3-fs-hadoop-1.19.1.jar — Gives Apache Flink learn and write entry to Amazon S3
• flink-sql-connector-hive-3.1.3_2.12-1.19.1.jar — Hive metastore connector for catalog compatibility
• hadoop-common-3.4.0.jar — Core Hadoop libraries required by Apache Iceberg
• flink-shaded-hadoop-2-uber-2.8.3-10.0.jar — Repackaged Hadoop dependencies that keep away from model conflicts with Apache Flink
• hadoop-hdfs-client-3.4.0.jar — Hadoop Distributed File System (HDFS) shopper libraries for file system operations
• flink-json-1.19.1.jar — JSON format help for Apache Flink
• hadoop-aws-3.4.0.jar — Hadoop integration with AWS companies
• hadoop-client-3.4.0.jar — Hadoop shopper libraries
• aws-java-sdk-bundle-1.12.261.jar — AWS SDK for authentication and repair entry

jars = [
    "flink-s3-fs-hadoop-1.19.1.jar",
    "flink-sql-connector-hive-3.1.3_2.12-1.19.1.jar",
    "hadoop-common-3.4.0.jar",
    "flink-shaded-hadoop-2-uber-2.8.3-10.0.jar",
    "iceberg-flink-runtime-1.19-1.6.1.jar",
    "iceberg-aws-bundle-1.6.1.jar",
    "hadoop-hdfs-client-3.4.0.jar",
    "flink-json-1.19.1.jar",
    "hadoop-aws-3.4.0.jar",
    "hadoop-client-3.4.0.jar",
    "aws-java-sdk-bundle-1.12.261.jar"
]

Technical implementation

The pattern code on this publish is accessible below the MIT-0 license.This part walks you thru constructing the streaming pipeline step-by-step. You create a single Python file, iceberg_streaming.py, with three features that run in sequence. Your predominant() perform calls them so as: arrange the Apache Flink surroundings, register the Information Catalog, then begin the streaming question.

Arrange your Apache Flink surroundings

To arrange your Apache Flink surroundings:

1. Obtain the required JAR recordsdata listed within the conditions part.
2. Place the JAR recordsdata in a lib listing in your undertaking folder.
3. Configure your HADOOP_CLASSPATH surroundings variable to level to the lib listing.
4. Create your streaming execution surroundings by including the next perform to iceberg_streaming.py:

def setup_environment():
    """Configure the Flink streaming runtime."""
    attempt:
        os.environ['HADOOP_CLASSPATH'] = os.path.be part of(os.getcwd(), 'lib', '*')
        env = StreamExecutionEnvironment.get_execution_environment()
        env.set_parallelism(1)
        settings = EnvironmentSettings.new_instance().in_streaming_mode().construct()
        t_env = StreamTableEnvironment.create(env, settings)
        return t_env
    besides Exception as e:
        print(f"Did not initialize Flink surroundings: {e}")
        elevate

5. Confirm your surroundings by working flink –model. If the command isn’t discovered, affirm that Apache Flink 1.19.1 is put in and that your PATH contains the Flink bin listing.

Configure AWS Glue Information Catalog

To attach your Flink software to Information Catalog:

1. Open your iceberg_streaming.py file.
2. Add the create_iceberg_source() perform proven within the following part.
3. Change the placeholder values together with your precise AWS sources earlier than working. These values are static configuration strings, not consumer enter — don’t assemble them from exterior or untrusted sources at runtime.
4. Save the file.

def create_iceberg_source(t_env):
    """Register the AWS Glue Information Catalog as an Iceberg catalog."""
    attempt:
        catalog_sql = """
        CREATE CATALOG glue_catalog WITH (
            'sort'='iceberg',
            'catalog-impl'='org.apache.iceberg.aws.glue.GlueCatalog',
            'warehouse'='s3://',
            'io-impl'='org.apache.iceberg.aws.s3.S3FileIO',
            'aws.area'='us-east-1',
            'hadoop-conf.fs.s3a.aws.credentials.supplier'=
                'com.amazonaws.auth.DefaultAWSCredentialsProviderChain',
            'hadoop-conf.fs.s3a.endpoint'='s3.amazonaws.com',
            'property-version'='1'
        )
        """
        t_env.execute_sql(catalog_sql)
        t_env.use_catalog("glue_catalog")
        t_env.use_database("streaming_db")
    besides Exception as e:
        print(f"Did not configure Iceberg catalog: {e}")
        elevate

Arrange streaming logic

This perform configures Apache Flink to observe your Apache Iceberg desk repeatedly and course of new data as they arrive. Checkpointing runs each 10 seconds to trace progress—if the job restarts, it resumes from the final checkpoint quite than reprocessing your entire desk.Discover the monitor-interval parameter, it controls how regularly Apache Flink checks for brand spanking new Apache Iceberg snapshots. A 3-second interval offers close to real-time processing however generates roughly 1,200 Amazon S3 LIST API calls per hour (at $0.005 per 1,000 requests, roughly $0.04/month per desk based mostly on pricing as of March 2026). For much less time-sensitive workloads, improve this to 30s to cut back API prices by 90%.Change customer_events with the title of your Apache Iceberg desk in Information Catalog:

def process_record(row):
    """Validate and course of every document from the stream."""
    attempt:
        if row is None:
            elevate ValueError("Obtained null row")
        required_fields = ["event_type", "timestamp"]
        for discipline in required_fields:
            if discipline not in row:
                elevate ValueError(f"Lacking required discipline: {discipline}")
        # Validate discipline varieties and content material
        if not isinstance(row.get("event_type"), str) or len(row["event_type"]) > 256:
            elevate ValueError("event_type have to be a string below 256 characters")
        if not isinstance(row.get("timestamp"), (str, int)):
            elevate ValueError("timestamp have to be a string or integer")
        # Change with your corporation logic
        print(f"Processing document: {row}")
    besides ValueError as e:
        print(f"Validation error for document {row}: {e}")
    besides Exception as e:
        print(f"Error processing document {row}: {e}")
def stream_data(t_env):
    """Begin the streaming question and course of outcomes."""
    attempt:
        configuration = t_env.get_config().get_configuration()
        configuration.set_string("desk.dynamic-table-options.enabled", "true")
        configuration.set_string("execution.checkpointing.interval", "10000")
        question = """
        SELECT * FROM customer_events /*+ OPTIONS(
            'streaming'='true',
            'monitor-interval'='3s',
            'desk.exec.iceberg.cell-based-snapshot'='true'
        ) */
        """
        table_result = t_env.execute_sql(question)
        with table_result.acquire() as outcomes:
            for row in outcomes:
                process_record(row)
    besides Exception as e:
        print(f"Streaming question failed: {e}")
        elevate

Placing it collectively

Your predominant() perform calls the three steps so as:

def predominant():
    attempt:
        t_env = setup_environment()
        create_iceberg_source(t_env)
        stream_data(t_env)
    besides Exception as e:
        print(f"Pipeline failed: {e}")
        elevate
if __name__ == "__main__":
    predominant()

Run the pipeline regionally:python iceberg_streaming.pyBundle the appliance and submit it to Amazon Managed Service for Apache Flink utilizing the console or the AWS Command Line Interface (AWS CLI).

Working in manufacturing

Transferring from a neighborhood check to a manufacturing deployment requires tuning 4 areas: efficiency, monitoring, value, and safety. This part covers the important thing choices for every.

Efficiency tuning

Decide your latency necessities earlier than tuning. For fraud detection, you want subsecond processing. For each day reporting dashboards, you’ll be able to tolerate minutes of delay.

Partition pruning reduces the quantity of knowledge scanned per question. Correct partitioning can considerably scale back question occasions for time collection knowledge partitioned by date. To implement, create your Apache Iceberg desk with partition columns (PARTITIONED BY (date_column) in your CREATE TABLE assertion), then embrace partition filters in your WHERE clause: WHERE date_column >= CURRENT_DATE - INTERVAL '7' DAY.

Parallel processing matches your knowledge quantity and throughput necessities. For many workloads below 10,000 data per second, a parallelism of 1–4 is enough. Scale up incrementally and monitor backpressure metrics (indicators that knowledge arrives quicker than your pipeline processes it, inflicting queuing) to seek out the correct setting.

Checkpoint tuning balances reliability and latency. Contemplate how a lot knowledge you’ll be able to afford to reprocess after a failure. In case you course of 1,000 data per second with 10-second checkpoints, a failure means reprocessing as much as 10,000 data. When that’s acceptable, 10 seconds works nicely. For quicker restoration or greater volumes, scale back to five seconds.

Useful resource allocation — Proper-size your Apache Flink cluster to keep away from over-provisioning. Monitor CPU and reminiscence utilization throughout your preliminary runs and alter process supervisor sources accordingly.

Monitoring

Configure your manufacturing deployment with the next checkpoint settings. These work nicely for average knowledge volumes (as much as 10,000 data per second), offering exactly-once processing semantics. Because of this the pipeline processes every document precisely as soon as, even when your software restarts. Regulate the checkpoint interval based mostly in your latency necessities. Add this to your setup_environment() perform after creating the desk surroundings.

config_dict = {
    "execution.checkpointing.interval": "30000",
    "execution.checkpointing.mode": "EXACTLY_ONCE",
    "execution.checkpointing.timeout": "600000",
    "state.backend": "filesystem",
    "state.checkpoints.dir": "s3:///checkpoints"
}

Use Amazon CloudWatch to trace checkpoint period, data processed per second, and backpressure metrics. A ten-second checkpoint interval means writing state to Amazon S3 360 occasions per hour. For a 1 MB state measurement, that’s roughly 8.6 GB per day in checkpoint storage—at Amazon S3 Customary pricing of $0.023/GB, roughly $0.20/day or $6/month per software based mostly on present pricing. If the checkpoint period exceeds 50% of your interval, improve the interval or add parallelism.

Value administration

Use Amazon S3 Clever-Tiering in your Apache Iceberg knowledge recordsdata, which generally have predictable entry patterns after preliminary processing. Configure Apache Iceberg’s desk expiration to mechanically clear up early snapshots. This could scale back storage prices by an estimated 20–30%, although your outcomes fluctuate relying on write frequency and retention insurance policies.

Proper-size your Apache Flink sources based mostly on precise throughput wants. Begin with a minimal configuration and scale up based mostly on noticed backpressure and checkpoint period metrics. Use Amazon Elastic Compute Cloud (Amazon EC2) Spot Situations the place workload interruptions are acceptable, for instance, in improvement and testing environments.

Set knowledge retention insurance policies on each your Apache Iceberg tables and checkpoint storage to keep away from storing knowledge longer than obligatory.

Safety

Safety is a shared duty between you and AWS. AWS is answerable for the safety of the cloud, together with the {hardware}, software program, networking, and amenities that run AWS companies. You’re answerable for safety within the cloud, configuring entry controls, encrypting knowledge, and managing your software safety. Apply these controls in precedence order.

AWS IAM roles — Use AWS IAM roles with least-privilege entry, scoped to particular sources. The next instance coverage restricts permissions to your knowledge lake bucket and AWS Glue catalog:

{
  "Model": "2012-10-17",
  "Assertion": [
    {
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:PutObject"],
      "Useful resource": "arn:aws:s3:::/*"
    },
    {
      "Impact": "Enable",
      "Motion": "s3:ListBucket",
      "Useful resource": "arn:aws:s3:::",
      "Situation": {
        "StringEquals": {
          "aws:SourceVpce": ""
        }
      }
    },
    {
      "Impact": "Enable",
      "Motion": ["glue:GetDatabase", "glue:GetTable"],
      "Useful resource": [
        "arn:aws:glue:us-east-1::catalog",
        "arn:aws:glue:us-east-1::database/streaming_db",
        "arn:aws:glue:us-east-1::table/streaming_db/*"
      ]
    },
    {
      "Impact": "Enable",
      "Motion": ["kms:Decrypt", "kms:GenerateDataKey"],
      "Useful resource": "arn:aws:kms:us-east-1::key/"
    }
  ]
}

Scoping permissions to particular Amazon S3 buckets, AWS Glue databases, and AWS Key Administration Service (AWS KMS) keys prohibit entry to solely the sources your pipeline requires. Evaluate IAM insurance policies quarterly utilizing the IAM Entry Analyzer to establish and take away unused permissions.

Encryption — Configure server-side encryption with AWS Key Administration Service (AWS KMS) buyer managed keys (SSE-KMS) in your Amazon S3 buckets. Utilizing buyer managed keys requires further overview out of your safety staff. Affirm your key administration insurance policies, rotation procedures, and entry controls earlier than implementation. Allow automated key rotation yearly. For encryption in transit, implement TLS by including a bucket coverage that denies non-HTTPS entry:

{
  "Impact": "Deny",
  "Principal": "*",
  "Motion": ["s3:GetObject", "s3:PutObject", "s3:ListBucket"],
  "Useful resource": [
    "arn:aws:s3:::/*",
    "arn:aws:s3:::"
  ],
  "Situation": {
    "Bool": { "aws:SecureTransport": "false" }
  }
}

Amazon S3 bucket hardening — Allow Block Public Entry in your buckets to stop unintentional public publicity:

aws s3api put-public-access-block 
  --bucket  
  --public-access-block-configuration 
  BlockPublicAcls=true,IgnorePublicAcls=true,BlockPublicPolicy=true,RestrictPublicBuckets=true

Allow versioning on buckets that retailer essential knowledge and checkpoints to guard in opposition to unintentional deletion. For manufacturing environments with delicate knowledge, contemplate enabling MFA Delete on versioned buckets. Allow S3 server entry logging to trace requests for safety auditing.

Amazon Digital Non-public Cloud (Amazon VPC) –Use Amazon VPC endpoints for personal communication between your Apache Flink cluster and AWS companies, eradicating public web routing by preserving visitors throughout the AWS community.

Entry logging – Allow AWS CloudTrail knowledge occasions to log Amazon S3 object-level API calls (GetObject, PutObject) and Information Catalog API calls. Retailer logs in a separate Amazon S3 bucket with restricted entry and allow log file integrity validation. Run common compliance checks utilizing AWS Config.

Operational practices

Arrange a steady integration and steady deployment (CI/CD) pipeline to automate deployment and testing. Use model management to trace schema and code adjustments. With Apache Iceberg’s schema evolution help, you’ll be able to add columns with out rewriting current knowledge recordsdata. Set up rollback procedures utilizing Apache Iceberg’s snapshot-based structure, so you’ll be able to roll again to a earlier desk state if a foul write corrupts your knowledge.

Troubleshooting

In case you run into points throughout setup or execution, use the next desk to diagnose frequent errors.

Clear up

To keep away from ongoing fees, delete the sources that you just created throughout this walkthrough.

1. Cease your Amazon Managed Service for Apache Flink software. Open the Amazon Managed Service for Apache Flink console, select your software title, select Cease, and make sure the motion. Or use the AWS CLI:

aws kinesisanalyticsv2 stop-application --application-name your-app-name

2. Delete the Amazon S3 buckets that you just created for knowledge storage and checkpoints. For directions, see Deleting a bucket within the Amazon S3 Consumer Information.
3. Take away the Apache Iceberg tables out of your Information Catalog.
4. Delete the AWS IAM roles and insurance policies created particularly for this walkthrough.
5. In case you created an Amazon VPC or Amazon VPC endpoints for testing, delete these sources.

Conclusion

Sustaining separate streaming and batch pipelines doubles your infrastructure prices, creates knowledge synchronization points, and provides operational complexity that slows your staff down. On this publish, you changed that dual-pipeline structure with a single system constructed on Apache Iceberg and Amazon Managed Service for Apache Flink. You configured a Flink surroundings with the required JAR dependencies, related it to Information Catalog, and carried out streaming queries that learn new data incrementally with exactly-once processing semantics. The identical knowledge, the identical storage layer, the identical schema—accessible to each your real-time and batch shoppers.

To increase this resolution, attempt these subsequent steps based mostly in your use case:

• In case you’re processing excessive volumes (>10,000 data/sec): Begin with partition pruning. Add PARTITIONED BY (date_column) to your desk definition, this sometimes reduces question occasions by 60–80%.
• In case you want manufacturing monitoring: Implement customized Amazon CloudWatch metrics. Observe checkpoint period, data processed per second, and backpressure to catch points earlier than they influence your pipeline.
• When you have variable workloads: Configure auto scaling in your Apache Flink cluster. See the Amazon Managed Service for Apache Flink Developer Information for detailed steerage.

Share your implementation expertise within the feedback, your use case, knowledge volumes, latency enhancements, and price reductions assist different readers calibrate their expectations. To get began, attempt the Amazon Managed Service for Apache Flink Developer Information and the Apache Iceberg documentation on the Apache Iceberg web site.

In regards to the authors