Stream real-time information into Apache Iceberg tables in Amazon S3 utilizing Amazon Knowledge Firehose

As companies generate extra information from quite a lot of sources, they want methods to successfully handle that information and use it for enterprise outcomes—equivalent to offering higher buyer experiences or lowering prices. We see these developments throughout many industries—on-line media and gaming firms offering suggestions and customised promoting, factories monitoring tools for upkeep and failures, theme parks offering wait instances for well-liked points of interest, and plenty of others.

To construct such purposes, engineering groups are more and more adopting two developments. First, they’re changing batch information processing pipelines with real-time streaming, so purposes can derive perception and take motion inside seconds as an alternative of ready for day by day or hourly batch alternate, remodel, and cargo (ETL) jobs. Second, as a result of conventional information warehousing approaches are unable to maintain up with the quantity, velocity, and number of information, engineering groups are constructing information lakes and adopting open information codecs equivalent to Parquet and Apache Iceberg to retailer their information. Iceberg brings the reliability and ease of SQL tables to Amazon Easy Storage Service (Amazon S3) information lakes. Through the use of Iceberg for storage, engineers can construct purposes utilizing completely different analytics and machine studying frameworks equivalent to Apache Spark, Apache Flink, Presto, Hive, or Impala, or AWS companies equivalent to Amazon SageMaker, Amazon Athena, AWS Glue, Amazon EMR, Amazon Managed Service for Apache Flink, or Amazon Redshift.

Iceberg is well-liked as a result of first, it’s broadly supported by completely different open-source frameworks and distributors. Second, it permits prospects to learn and write information concurrently utilizing completely different frameworks. For instance, you may write some data utilizing a batch ETL Spark job and different information from a Flink software on the similar time and into the identical desk. Third, it permits situations equivalent to time journey and rollback, so you may run SQL queries on a point-in-time snapshot of your information, or rollback information to a beforehand identified good model. Fourth, it helps schema evolution, so when your purposes evolve, you may add new columns to your tables with out having to rewrite information or change current purposes. To be taught extra, see Apache Iceberg.

On this put up, we focus on how one can ship real-time information streams into Iceberg tables on Amazon S3 through the use of Amazon Knowledge Firehose. Amazon Knowledge Firehose simplifies the method of streaming information by permitting customers to configure a supply stream, choose an information supply, and set Iceberg tables because the vacation spot. As soon as arrange, the Firehose stream is able to ship information. Firehose is built-in with over 20 AWS companies, so you may ship real-time information from Amazon Kinesis Knowledge Streams, Amazon Managed Streaming for Apache Kafka, Amazon CloudWatch Logs, AWS Web of Issues (AWS IoT), AWS WAF, Amazon Community Firewall Logs, or out of your customized purposes (by invoking the Firehose API) into Iceberg tables. It’s price efficient as a result of Firehose is serverless, you solely pay for the information despatched and written to your Iceberg tables. You don’t should provision something or pay something when your streams are idle throughout nights, weekends, or different non-use hours.

Firehose additionally simplifies organising and operating superior situations. For instance, if you wish to route information to completely different Iceberg tables to have information isolation or higher question efficiency, you may arrange a stream to robotically route data into completely different tables based mostly on what’s in your incoming information and distribute data from a single stream into dozens of Iceberg tables. Firehose robotically scales—so that you don’t should plan for the way a lot information goes into which desk—and has built-in mechanisms to deal with supply failures and assure precisely as soon as supply. Firehose helps updating and deleting data in a desk based mostly on the incoming information stream, so you may assist situations equivalent to GDPR and right-to-forget laws. As a result of Firehose is absolutely appropriate with Iceberg, you may write information utilizing it and concurrently use different purposes to learn and write to the identical tables. Firehose integrates with the AWS Glue Knowledge Catalog, so you should use options in AWS Glue equivalent to managed compaction for Iceberg tables.

Within the following sections, you’ll discover ways to arrange Firehose to ship real-time streams into Iceberg tables to handle 4 completely different situations:

1. Ship information from a stream right into a single Iceberg desk and insert all incoming data.
2. Ship information from a stream right into a single Iceberg desk and carry out file inserts, updates, and deletes.
3. Route data to completely different tables based mostly on the content material of the incoming information by specifying a JSON Question expression.
4. Route data to completely different tables based mostly on the content material of the incoming information through the use of a Lambda operate.

Additionally, you will discover ways to question the information you have got delivered to Iceberg tables utilizing an ordinary SQL question in Amazon Athena. All the AWS companies utilized in these examples are serverless, so that you don’t should provision and handle any infrastructure.

Answer overview

The next diagram illustrates the structure.

In our examples, we use Kinesis Knowledge Generator, a pattern software to generate and publish information streams to Firehose. You can even arrange Firehose to make use of different information sources on your real-time streams. We arrange Firehose to ship the stream into Iceberg tables within the Knowledge Catalog.

Walkthrough

This put up consists of an AWS CloudFormation template for a fast setup. You may overview and customise it to fit your wants. The template performs the next operations:

• Creates a Knowledge Catalog database for the vacation spot Iceberg tables
• Creates 4 tables within the AWS Glue database which might be configured to make use of the Apache Iceberg format
• Specifies the S3 bucket places for the vacation spot tables
• Creates a Lambda operate (non-obligatory)
• Units up an AWS Id and Entry Administration (IAM) function for Firehose
• Creates assets for Kinesis Knowledge Generator

Stipulations

For this walkthrough, it’s best to have the next conditions:

• An AWS account. Should you don’t have an account, you may create one.

Deploy the answer

Step one is to deploy the required assets into your AWS setting through the use of a CloudFormation template.

1. Sign up to the AWS Administration Console for CloudFormation.
2. Select Launch Stack.
   
3. Select Subsequent.
4. Go away the stack title as Firehose-Iceberg-Stack, and within the parameters, enter the username and password that you just need to use for accessing Kinesis Knowledge Generator.
   
5. Go to the underside of the web page and choose I acknowledge that AWS CloudFormation may create IAM assets and select Subsequent.
   
   
6. Assessment the deployment and select Submit.
   

The stack can take 5–10 minutes to finish, after which you’ll be able to view the deployed stack on the CloudFormation console. The next determine reveals the deployed Firehose-Iceberg-stack particulars.

Earlier than you arrange Firehose to ship streams, you have to create the vacation spot tables within the Knowledge Catalog. For the examples mentioned right here, we use the CloudFormation template to robotically create the tables used within the examples. In your customized purposes, you may create your tables utilizing CloudFormation, or through the use of DDL instructions in Athena or Glue. The next is the DDL command for making a desk utilized in our instance:

CREATE TABLE firehose_iceberg_db.firehose_events_1 (
kind struct,
customer_id string,
event_timestamp timestamp,
area string)
LOCATION 's3://firehose-demo-iceberg--/iceberg/events_1'
TBLPROPERTIES (
'table_type'='iceberg',
'write_compression'='zstd'
);

Additionally notice that the 4 tables that we use within the examples have the identical schema, however you may have tables with completely different schemas in your software.

Use case 1: Ship information from a stream right into a single Iceberg desk and insert all incoming data

Now that you’ve arrange the supply on your information stream and the vacation spot tables, you’re able to arrange Firehose to ship streams into the Iceberg tables.

Create a Firehose stream:

1. Go to the Knowledge Firehose console and select Create Firehose stream.
   
2. Choose Direct PUT because the Supply and Apache Iceberg Tables because the Vacation spot.

This instance makes use of Direct PUT because the supply, however the identical steps might be utilized for different Firehose sources equivalent to Kinesis Knowledge Streams, and Amazon MSK.

3. For the Firehose stream title, enter firehose-iceberg-events-1.
   
4. In Vacation spot settings, allow Inline parsing for routing info. As a result of all data from the stream are inserted right into a single vacation spot desk, you specify a vacation spot database and desk. By default, Firehose inserts all incoming data into the desired vacation spot desk.
   1. Database expression: “firehose_iceberg_db”
   2. Desk expression: “firehose_events_1”

Embrace double citation marks to make use of the literal worth for the database and desk title. If you don’t use double quotations marks, Firehose assumes that this can be a JSON Question expression and can try to parse the expression when processing your stream and fail.

5. Go to Buffer hints and scale back the Buffer dimension to 1 MiB and the Buffer interval to 60 You may nice tune these settings on your software.
   
6. For Backup settings:
   • Choose the S3 bucket created by the CloudFormation template. It has the next construction: s3://firehose-demo-iceberg--
   • For error output prefix enter: error/events-1/

   

7. In Superior settings, allow CloudWatch error logging, and in Current IAM roles, choose the function that begins with Firehose-Iceberg-Stack-FirehoseIamRole-*, created by the CloudFormation template.
8. Select Create Firehose stream.
   

Generate streaming information:

Use Kinesis Knowledge Generator to publish information data into your Firehose stream.

1. Go to the CloudFormation stack, choose the Nested stack for the generator, and select Outputs.
   
2. Select the KinesisDataGenerator URL and enter the credentials that you just outlined when deploying the CloudFormation stack.
   
3. Choose the AWS Area the place you deployed the CloudFormation stack and choose your Firehose stream.
   
4. For template, exchange the values on the display screen with the next:

   {
   "kind": {
   "system": "{{random.arrayElement(["mobile", "desktop", "tablet"])}}",
   "occasion": "{{random.arrayElement(["firehose_events_1", "firehose_events_2"])}}",
   "motion": "replace"
   },
   "customer_id": "{{random.quantity({ "min": 1, "max": 1500})}}",
   "event_timestamp": "{{date.now("YYYY-MM-DDTHH:mm:ss.SSS")}}",
   "area": "{{random.arrayElement(["pdx", "nyc"])}}"
   }

   

5. Earlier than sending information, select Take a look at template to see an instance payload.
6. Select Ship information.
   

Querying with Athena:

You may question the information you’ve written to your Iceberg tables utilizing completely different processing engines equivalent to Apache Spark, Apache Flink, or Trino. On this instance, we are going to present you the way you should use Athena to question information that you just’ve written to Iceberg tables.

1. Go to the Athena console.
2. Configure a Location of question consequence. You should utilize the identical S3 bucket for this however add a suffix on the finish.

   s3://firehose-demo-iceberg--/athena/

   

3. Within the question editor, in Tables and views, choose the choices button subsequent to firehose_events_1 and choose Preview Desk.
   

It is best to have the ability to see information within the Apache Iceberg tables through the use of Athena.

With that, you ‘ve delivered information streams into an Apache Iceberg desk utilizing Firehose and run a SQL question in opposition to your information.

Now let’s discover the opposite situations. We’ll comply with the identical process as earlier than for creating the Firehose stream and querying Iceberg tables with Amazon Athena.

Use case 2: Ship information from a stream right into a single Iceberg desk and carry out file inserts, updates, and deletes

One of many benefits of utilizing Apache Iceberg is that it means that you can carry out row-level operations equivalent to updates and deletes on tables in an information lake. Firehose might be set as much as robotically apply file replace and delete operations in your Iceberg tables.

Issues to know:

• While you apply an replace or delete operation by Firehose, the information in Amazon S3 isn’t really deleted. As a substitute, a marker file is written in keeping with the Apache Iceberg format specification to point that the file is up to date or deleted, so subsequent learn and write operations get the most recent file. If you wish to purge (take away the underlying information from Amazon S3) the deleted data, you should use instruments developed for purging data in Apache Iceberg.
• Should you try to replace a file utilizing Firehose and the underlying file doesn’t exist already within the vacation spot desk, Firehose will insert the file as a brand new row.

Create a Firehose stream:

1. Go to the Amazon Knowledge Firehose console.
2. Select Create Firehose stream.
3. For Supply, choose Direct PUT. For Vacation spot choose Apache Iceberg Tables.
4. For the Firehose stream title, enter firehose-iceberg-events-2.
5. Within the e, allow inline parsing for routing info and supply the required values as static values for Database expression and Desk expression. Since you need to have the ability to replace data, you additionally have to specify the Operation expression.
   1. Database expression: “firehose_iceberg_db”
   2. Desk expression: “firehose_events_2”
   3. Operation expression: “replace”

Embrace double citation marks to make use of the literal worth for the database and desk title. If you don’t use double quotations marks, Firehose assumes that this can be a JSON Question expression and can try to parse the expression when processing your stream and fail.

6. Since you need to carry out replace and delete operations, that you must present the columns within the vacation spot desk that can be used as distinctive keys to determine the file within the vacation spot to be up to date or deleted.
   • For DestinationDatabaseName: “firehose_iceberg_db“
   • For DestinationTableName: “firehose_events_2”
   • In UniqueKeys, exchange the prevailing worth with: “customer_id”

   

7. Change the Buffer hints to 1 MiB and 60
8. In Backup settings, choose the identical bucket from the stack, however enter the next within the error output prefix:
9. In Superior settings, allow CloudWatch Error logging and choose the prevailing function of the stack and create the brand new Firehose stream.

Use Kinesis Knowledge Generator to publish data into your Firehose stream. You may have to refresh the web page or change areas in order that it refreshes and reveals the newly created supply stream.

Don’t make any modifications to the template and begin sending information to the firehose-iceberg-events-2 stream.

Run the next question in Athena to see information within the firehose_events_2 desk. Be aware that you would be able to ship up to date data for a similar distinctive key (similar customer_id worth) into your Firehose stream, and Firehose robotically applies file updates within the vacation spot desk. Thus, once you question information in Athena, you will note just one file for every distinctive worth of customer_id, even when you’ve got despatched a number of updates into your stream.

SELECT customer_id, rely(*) 
FROM "firehose_iceberg_db"."firehose_events_2" 
GROUP BY customer_id LIMIT 10;

Use case 3: Route data to completely different tables based mostly on the content material of the incoming information by specifying a JSON Question expression

Till now, you offered the routing and operation info as static values to carry out operations on a single desk. Nevertheless, you may specify JSON Question expressions to outline how Firehose ought to retrieve the vacation spot database, vacation spot desk, and operation out of your incoming information stream, and accordingly route the file and carry out the corresponding operation. Based mostly in your specification, Firehose robotically routes and delivers every file into the suitable vacation spot desk and applies the corresponding operation.

Create a Firehose stream:

1. Return to the Amazon Knowledge Firehose console.
2. Select Create Firehose Stream.
3. For Supply, choose Direct PUT. For Vacation spot, choose Apache Iceberg Tables.
4. For the Firehose stream title, enter firehose-iceberg-events-3.
   
5. In Vacation spot settings, allow Inline parsing for routing info.
   • For Database expression, present the identical worth as earlier than as a static string: “firehose_iceberg_db”
   • For Desk expression, retrieve this worth from the nested incoming file utilizing JSON Question.
   • For Operation expression, we may even retrieve this worth from our nested file utilizing JSON Question.

When you have the next incoming occasions with completely different occasion values, With the previous JSON Question expressions, Firehose will parse and get “firehose_event_3” or “firehose_event_4” because the desk names, and “replace” because the supposed operation from the incoming data.

{ "kind": {   "system": "pill",  
"occasion": "firehose_events_3",   "motion": "replace" },
"customer_id": "112", "event_timestamp": "2024-10-02T15:46:52.901",
"area": "pdx"}
{ "kind": {   "system": "pill",  
"occasion": "firehose_events_4",   "motion": "replace" },
"customer_id": "112", "event_timestamp": "2024-10-02T15:46:52.901",
"area": "pdx"}

6. As a result of that is an replace operation, that you must configure distinctive keys for every desk. Additionally, since you need to ship data to a number of Iceberg tables, that you must present configurations for every of the 2 vacation spot tables that data might be written to.
   
7. Change the Buffer hints to 1 MiB and 60
8. In Backup settings, choose the identical bucket from the stack, however within the error output prefix enter the next:
9. In Superior settings, choose the prevailing IAM function created by the CloudFormation stack and create the brand new Firehose stream.
10. In Kinesis Knowledge Generator, refresh the web page and choose the newly created Firehose stream: firehose-iceberg-events-3

Should you question the firehose_events_3 and firehose_events_4 tables utilizing Athena, it’s best to discover the information routed to proper tables by Firehose utilizing the routing info retrieved utilizing JSON Question expressions.

Desk beneath displaying  occasions with occasion “firehose_events_3”

The next determine reveals Firehose Occasions Desk 4.

Use Case 4: Route data to completely different tables based mostly on the content material of the incoming information through the use of a Lambda operate

There may be situations the place routing info isn’t available within the enter file. You may need to parse and course of incoming data or carry out a lookup to find out the place to ship the file and whether or not to carry out an replace or delete operation. For such situations, you should use a Lambda operate to generate the routing info and operation specification. Firehose robotically invokes your Lambda operate for a batch of data (with a configurable batch dimension). You may course of incoming data in your Lambda operate and supply the routing info and operation within the output of the operate. To be taught extra about learn how to course of Firehose data utilizing Lambda, see Rework supply information in Amazon Knowledge Firehose. After executing your Lambda operate, Firehose appears to be like for routing info and operations within the metadata fields (within the following format) offered by your Lambda operate.

    "metadata":{
        "otfMetadata":{
            "destinationTableName":"firehose_iceberg_db",
            "destinationDatabaseName":"firehose_events_*",
            "operation":"insert"
        }

So, on this use case, you’ll discover how one can create customized routing guidelines based mostly on different values of your data. Particularly, for this use case, you’ll route all data with a price for Area of ‘pdx’ to desk 3 and all data with a area worth of ‘nyc’ to desk 4.

The CloudFormation template has already created the customized processing Lambda operate for you, which has the next code:

import base64
import json
print('Loading operate')

def lambda_handler(occasion, context):
    firehose_records_output = {'data': []}

    for file in occasion['records']:
        payload = base64.b64decode(file['data']).decode('utf-8')
        # Course of the payload based mostly on area
        payload_json = json.masses(payload)
        area = payload_json.get('area', '')
        firehose_record_output = {}
        if area == 'pdx':
            firehose_record_output['metadata'] = {
                'otfMetadata': {
                    'destinationDatabaseName': 'firehose_iceberg_db',
                    'destinationTableName': 'firehose_events_3',
                    'operation': 'insert'
                }
            }
        elif area == 'nyc':
            firehose_record_output['metadata'] = {
                'otfMetadata': {
                    'destinationDatabaseName': 'firehose_iceberg_db',
                    'destinationTableName': 'firehose_events_4',
                    'operation': 'insert'
                }
            }

        # Create output with correct file ID, output information, consequence, and metadata
        firehose_record_output['recordId'] = file['recordId']
        firehose_record_output['result'] = 'Okay'
        firehose_record_output['data'] = base64.b64encode(payload.encode('utf-8'))
        firehose_records_output['records'].append(firehose_record_output)

    return firehose_records_output

Configure the Firehose stream:

1. Return to the Knowledge Firehose console.
2. Select Create Firehose stream.
3. For Supply, choose Direct PUT. For Vacation spot, choose Apache Iceberg Tables.
4. For the Firehose stream title, enter firehose-iceberg-events-4.
   
5. In Rework data, choose Activate information transformation.
6. Browse and choose the operate created by the CloudFormation stack:
   • Firehose-Iceberg-Stack-FirehoseProcessingLambda-*.
   • For Model choose $LATEST.
     
7. You may go away the Vacation spot Settings as default as a result of the Lambda operate will present the required metadata for routing.
8. Change the Buffer hints to 1 MiB and 60 seconds.
9. In Backup settings, choose the identical S3 bucket from the stack, however within the error output prefix, enter the next:
10. In Superior settings, choose the prevailing function of the stack and create the brand new Firehose stream.
11. In Kinesis Knowledge Generator, refresh the web page and choose the newly created firehose stream: firehose-iceberg-events-4.

Should you run the next question, you will note that the final data that have been inserted into the desk are solely within the Area of ‘nyc’.

SELECT * FROM "firehose_iceberg_db"."firehose_events_4" 
order by event_timestamp desc 
restrict 10;

Issues and limitations

Earlier than utilizing Knowledge Firehose with Apache Iceberg, it’s vital to pay attention to concerns and limitations. For extra info, see Issues and limitations.

Clear up

To keep away from future fees, delete the assets you created in AWS Glue, Knowledge Catalog, and the S3 bucket used for storage.

Conclusion

It’s simple to arrange Firehose streams to ship streaming data into Apache Iceberg tables in Amazon S3. We hope that this put up helps you get began with constructing some wonderful purposes with out having to fret about writing and managing complicated software code or having to handle infrastructure.

To be taught extra about utilizing Amazon Knowledge Firehose with Apache Iceberg, see the Firehose Developer Information or strive the Immersion day workshop.

Concerning the authors

Diego Garcia Garcia is a Specialist SA Supervisor for Analytics at AWS. His experience spans throughout Amazon’s analytics companies, with a specific deal with real-time information processing and superior analytics architectures. Diego leads a crew of specialist options architects throughout EMEA, collaborating carefully with prospects spanning throughout a number of industries and geographies to design and implement options to their information analytics challenges.

Francisco Morillo is a Streaming Options Architect at AWS. Francisco works with AWS prospects, serving to them design real-time analytics architectures utilizing AWS companies, supporting Amazon Managed Streaming for Apache Kafka (Amazon MSK) and Amazon Managed Service for Apache Flink.

Phaneendra Vuliyaragoli is a Product Administration Lead for Amazon Knowledge Firehose at AWS. On this function, Phaneendra leads the product and go-to-market technique for Amazon Knowledge Firehose.