How Cloudinary reworked their petabyte scale streaming knowledge lake with Apache Iceberg and AWS Analytics

This submit is co-written with Amit Gilad, Alex Dickman and Itay Takersman from Cloudinary. 

Enterprises and organizations throughout the globe wish to harness the facility of knowledge to make higher choices by placing knowledge on the heart of each decision-making course of. Knowledge-driven choices result in more practical responses to surprising occasions, enhance innovation and permit organizations to create higher experiences for his or her prospects. Nonetheless, all through historical past, knowledge providers have held dominion over their prospects’ knowledge. Regardless of the potential separation of storage and compute by way of structure, they’re typically successfully fused collectively. This amalgamation empowers distributors with authority over a various vary of workloads by advantage of proudly owning the information. This authority extends throughout realms akin to enterprise intelligence, knowledge engineering, and machine studying thus limiting the instruments and capabilities that can be utilized.

The panorama of knowledge know-how is swiftly advancing, pushed ceaselessly by tasks led by the open supply neighborhood generally and the Apache basis particularly. This evolving open supply panorama permits prospects full management over knowledge storage, processing engines and permissions increasing the array of accessible choices considerably. This method additionally encourages distributors to compete primarily based on the worth they supply to companies, somewhat than counting on potential fusing of storage and compute. This fosters a aggressive atmosphere that prioritizes buyer acquisition and prompts distributors to distinguish themselves by way of distinctive options and choices that cater on to the particular wants and preferences of their clientele.

A contemporary knowledge technique redefines and allows sharing knowledge throughout the enterprise and permits for each studying and writing of a singular occasion of the information utilizing an open desk format. The open desk format accelerates corporations’ adoption of a contemporary knowledge technique as a result of it permits them to make use of varied instruments on high of a single copy of the information.

Cloudinary is a cloud-based media administration platform that gives a complete set of instruments and providers for managing, optimizing, and delivering photos, movies, and different media property on web sites and cell functions. It’s extensively utilized by builders, content material creators, and companies to streamline their media workflows, improve person experiences, and optimize content material supply.

On this weblog submit, we dive into completely different knowledge elements and the way Cloudinary breaks the 2 issues of vendor locking and price environment friendly knowledge analytics by utilizing Apache Iceberg, Amazon Easy Storage Service (Amazon S3), Amazon Athena, Amazon EMR, and AWS Glue.

Brief overview of Cloudinary’s infrastructure

Cloudinary infrastructure handles over 20 billion requests each day with each request producing occasion logs. Numerous knowledge pipelines course of these logs, storing petabytes (PBs) of knowledge monthly, which after processing knowledge saved on Amazon S3, are then saved in Snowflake Knowledge Cloud. These datasets function a vital useful resource for Cloudinary inner groups and knowledge science teams to permit detailed analytics and superior use circumstances.

Till just lately, this knowledge was largely ready by automated processes and aggregated into outcomes tables, utilized by only some inner groups. Cloudinary struggled to make use of this knowledge for extra groups who had extra on-line, actual time, lower-granularity, dynamic utilization necessities. Making petabytes of knowledge accessible for ad-hoc reviews grew to become a problem as question time elevated and prices skyrocketed together with rising compute useful resource necessities. Cloudinary knowledge retention for the particular analytical knowledge mentioned on this submit was outlined as 30 days. Nonetheless, new use circumstances drove the necessity for elevated retention, which might have led to considerably greater value.

The information is flowing from Cloudinary log suppliers into recordsdata written into Amazon S3 and notified by way of occasions pushed to Amazon Easy Queue Service (Amazon SQS). These SQS occasions are ingested by a Spark software operating in Amazon EMR Spark, which parses and enriches the information. The processed logs are written in Apache Parquet format again to Amazon S3 after which routinely loaded to a Snowflake desk utilizing Snowpipe.

Why Cloudinary selected Apache Iceberg

Apache Iceberg is a high-performance desk format for large analytic workloads. Apache Iceberg brings the reliability and ease of SQL tables to large knowledge, whereas making it potential for processing engines akin to Apache Spark, Trino, Apache Flink, Presto, Apache Hive, and Impala to soundly work with the identical tables on the similar time.

An answer primarily based on Apache Iceberg encompasses full knowledge administration, that includes easy built-in desk optimization capabilities inside an present storage answer. These capabilities, together with the flexibility to make use of a number of engines on high of a singular occasion of knowledge, helps keep away from the necessity for knowledge motion between varied options.

Whereas exploring the assorted controls and choices in configuring Apache Iceberg, Cloudinary needed to adapt its knowledge to make use of AWS Glue Knowledge Catalog, in addition to transfer a big quantity of knowledge to Apache Iceberg on Amazon S3. At this level it grew to become clear that prices can be considerably decreased, and whereas it had been a key issue because the planning section, it was now potential to get concrete numbers. One instance is that Cloudinary was now in a position to retailer 6 months of knowledge for a similar storage value that was beforehand paid for storing 1 month of knowledge. This value saving was achieved by utilizing Amazon S3 storage tiers in addition to improved compression (Zstandard), additional enhanced by the truth that Parquet recordsdata had been sorted.

Since Apache Iceberg is effectively supported by AWS knowledge providers and Cloudinary was already utilizing Spark on Amazon EMR, they might combine writing to Knowledge Catalog and begin a further Spark cluster to deal with knowledge upkeep and compaction. As exploration continued with Apache Iceberg, some fascinating efficiency metrics had been discovered. For instance, for sure queries, Athena runtime was 2x–4x sooner than Snowflake.

Integration of Apache Iceberg

The mixing of Apache Iceberg was executed earlier than loading knowledge to Snowflake. The information is written to an Iceberg desk utilizing Apache Parquet knowledge format and AWS Glue as the information catalog. As well as, a Spark software on Amazon EMR runs within the background dealing with compaction of the Parquet recordsdata to optimum measurement for querying by way of varied instruments akin to Athena, Trino operating on high of EMR, and Snowflake.

Challenges confronted

Cloudinary confronted a number of challenges whereas constructing its petabyte-scale knowledge lake, together with:

• Figuring out optimum desk partitioning
• Optimizing ingestion
• Fixing the small recordsdata drawback to enhance question efficiency
• Affordably sustaining Apache Iceberg tables
• Selecting the best question engine

On this part, we describe every of those challenges and the options carried out to handle them. Lots of the assessments to examine efficiency and volumes of knowledge scanned have used Athena as a result of it offers a easy to make use of, absolutely serverless, value efficient, interface with out the necessity to setup infrastructure.

Figuring out optimum desk partitioning

Apache Iceberg makes partitioning simpler for the person by implementing hidden partitioning. Reasonably than forcing the person to produce a separate partition filter at question time, Iceberg tables might be configured to map common columns to the partition keys. Customers don’t want to take care of partition columns and even perceive the bodily desk structure to get quick and correct question outcomes.

Iceberg has a number of partitioning choices. One instance is when partitioning timestamps, which might be executed by 12 months, month, day, and hour. Iceberg retains monitor of the connection between a column worth and its partition with out requiring further columns. Iceberg may also partition categorical column values by id, hash buckets, or truncation. As well as, Iceberg partitioning is user-friendly as a result of it additionally permits partition layouts to evolve over time with out breaking pre-written queries. For instance, when utilizing each day partitions and the question sample modifications over time to be primarily based on hours, it’s potential to evolve the partitions to hourly ones, thus making queries extra environment friendly. When evolving such a partition definition, the information within the desk previous to the change is unaffected, as is its metadata. Solely knowledge that’s written to the desk after the evolution is partitioned with the brand new definition, and the metadata for this new set of knowledge is stored individually. When querying, every partition structure’s respective metadata is used to establish the recordsdata that must be accessed; that is referred to as split-planning. Break up-planning is one among many Iceberg options which might be made potential as a result of desk metadata, which creates a separation between the bodily and the logical storage. This idea makes Iceberg extraordinarily versatile.

Figuring out the right partitioning is vital when working with giant knowledge units as a result of it impacts question efficiency and the quantity of knowledge being scanned. As a result of this migration was from present tables from Snowflake native storage to Iceberg, it was essential to check and supply an answer with the identical or higher efficiency for the present workload and forms of queries.

These assessments had been potential attributable to Apache Iceberg’s:

1. Hidden partitions
2. Partition transformations
3. Partition evolution

These allowed altering desk partitions and testing which technique works finest with out knowledge rewrite.

Listed here are a couple of partitioning methods that had been examined:

1. PARTITIONED BY (days(day), customer_id)
2. PARTITIONED BY (days(day), hour(timestamp))
3. PARTITIONED BY (days(day), bucket(N, customer_id))
4. PARTITIONED BY (days(day))

Every partitioning technique that was reviewed generated considerably completely different outcomes each throughout writing in addition to throughout question time. After cautious outcomes evaluation, Cloudinary determined to partition the information by day and mix it with sorting, which permits them to kind knowledge inside partitions as can be elaborated within the compaction part.

Optimizing ingestion

Cloudinary receives billions of occasions in recordsdata from its suppliers in varied codecs and sizes and shops these on Amazon S3, leading to terabytes of knowledge processed and saved every single day.

As a result of the information doesn’t are available in a constant method and it’s not potential to foretell the incoming price and file measurement of the information, it was essential to discover a means of protecting value down whereas sustaining excessive throughput.

This was achieved by utilizing EventBridge to push every file acquired into Amazon SQS, the place it was processed utilizing Spark operating on Amazon EMR in batches. This allowed processing the incoming knowledge at excessive throughput and scale clusters in keeping with queue measurement whereas protecting prices down.

Instance of fetching 100 messages (recordsdata) from Amazon SQS with Spark:

var consumer = AmazonSQSClientBuilder.commonplace().withRegion("us-east-1").construct()
var getMessageBatch: Iterable[Message] = DistributedSQSReceiver.consumer.receiveMessage(new ReceiveMessageRequest().withQueueUrl(queueUrl).withMaxNumberOfMessages(10)).getMessages.asScala
sparkSession.sparkContext.parallelize(10) .map(_ => getMessageBatch) .acquire().flatMap(_.toList) .toList

When coping with a excessive knowledge ingestion price for a selected partition prefix, Amazon S3 would possibly doubtlessly throttle requests and return a 503 standing code (service unavailable). To deal with this state of affairs, Cloudinary used an Iceberg desk property referred to as write.object-storage.enabled, which includes a hash prefix into the saved Amazon S3 object path. This method was deemed environment friendly and successfully mitigated Amazon S3 throttling issues.

Fixing the small file drawback and bettering question efficiency

In fashionable knowledge architectures, stream processing engines akin to Amazon EMR are sometimes used to ingest steady streams of knowledge into knowledge lakes utilizing Apache Iceberg. Streaming ingestion to Iceberg tables can endure from two issues:

• It generates many small recordsdata that result in longer question planning, which in flip can influence learn efficiency.
• Poor knowledge clustering, which might make file pruning much less efficient. This usually happens within the streaming course of when there may be inadequate new knowledge to generate optimum file sizes for studying, akin to 512 MB.

As a result of partition is a key issue within the variety of recordsdata produced and Cloudinary’s knowledge is time primarily based and most queries use a time filter, it was determined to handle the optimization of our knowledge lake in a number of methods.

First, Cloudinary set all the mandatory configurations that helped scale back the variety of recordsdata whereas appending knowledge within the desk by setting write.target-file-size-bytes, which permits defining the default goal file measurement. Setting spark.sql.shuffle.partitions in Spark can scale back the variety of output recordsdata by controlling the variety of partitions used throughout shuffle operations, which impacts how knowledge is distributed throughout duties, consequently minimizing the variety of output recordsdata generated after transformations or aggregations.

As a result of the above method solely addressed the small file drawback however didn’t get rid of it solely, Cloudinary used one other functionality of Apache Iceberg that may compact knowledge recordsdata in parallel utilizing Spark with the rewriteDataFiles motion. This motion combines small recordsdata into bigger recordsdata to scale back metadata overhead and reduce the quantity of Amazon S3 GetObject API operation utilization.

Right here is the place it will possibly get difficult. When operating compaction, Cloudinary wanted to decide on which technique to use out of the three that Apache Iceberg affords; every one having its personal benefits and drawbacks:

1. Binpack – merely rewrites smaller recordsdata to a goal measurement
2. Type – knowledge sorting primarily based on completely different columns
3. Z-order – a way to colocate associated knowledge in the identical set of recordsdata

At first, the Binpack compaction technique was evaluated. This technique works quickest and combines small recordsdata collectively to succeed in the goal file measurement outlined and after operating it a big enchancment in question efficiency was noticed.

As talked about beforehand, knowledge was partitioned by day and most queries ran on a selected time vary. As a result of knowledge comes from exterior distributors and generally arrives late, it was seen that when operating queries on compacted days, loads of knowledge was being scanned, as a result of the particular time vary might reside throughout many recordsdata. The question engine (Athena, Snowflake, and Trino with Amazon EMR) wanted to scan all the partition to fetch solely the related rows.

To extend question efficiency even additional, Cloudinary determined to alter the compaction course of to make use of kind, so now knowledge is partitioned by day and sorted by requested_at (timestamp when the motion occurred) and buyer ID.

This technique is costlier for compaction as a result of it must shuffle the information in an effort to kind it. Nonetheless, after adopting this type technique, two issues had been noticeable: the identical queries that ran prior to now scanned round 50 % much less knowledge, and question run time was improved by 30 % to 50 %.

Affordably sustaining Apache Iceberg tables

Sustaining Apache Iceberg tables is essential for optimizing efficiency, decreasing storage prices, and guaranteeing knowledge integrity. Iceberg offers a number of upkeep operations to maintain your tables in fine condition. By incorporating these operations Cloudinary had been in a position to cost-effectively handle their Iceberg tables.

Expire snapshots

Every write to an Iceberg desk creates a brand new snapshot, or model, of a desk. Snapshots can be utilized for time-travel queries, or the desk might be rolled again to any legitimate snapshot.

Frequently expiring snapshots is really useful to delete knowledge recordsdata which might be now not wanted and to maintain the dimensions of desk metadata small. Cloudinary determined to retain snapshots for as much as 7 days to permit simpler troubleshooting and dealing with of corrupted knowledge which generally arrives from exterior sources and aren’t recognized upon arrival. SparkActions.get().expireSnapshots(iceTable).expireOlderThan(TimeUnit.DAYS.toMillis(7)).execute()

Take away outdated metadata recordsdata

Iceberg retains monitor of desk metadata utilizing JSON recordsdata. Every change to a desk produces a brand new metadata file to offer atomicity.

Previous metadata recordsdata are stored for historical past by default. Tables with frequent commits, like these written by streaming jobs, would possibly must recurrently clear metadata recordsdata.

Configuring the next properties will make it possible for solely the newest ten metadata recordsdata are stored and something older is deleted.

write.metadata.delete-after-commit.enabled=true 
write.metadata.previous-versions-max=10

Delete orphan recordsdata

In Spark and different distributed processing engines, when duties or jobs fail, they could depart behind recordsdata that aren’t accounted for within the desk metadata. Furthermore, in sure situations, the usual snapshot expiration course of would possibly fail to establish recordsdata which might be now not essential and never delete them.

Apache Iceberg affords a deleteOrphanFiles motion that can care for unreferenced recordsdata. This motion would possibly take a very long time to finish if there are a lot of recordsdata within the knowledge and metadata directories. A metadata or knowledge file is taken into account orphan if it isn’t reachable by any legitimate snapshot. The set of precise recordsdata is constructed by itemizing the underlying storage utilizing the Amazon S3 ListObjects operation, which makes this operation costly. It’s really useful to run this operation periodically to keep away from elevated storage utilization; nevertheless, too frequent runs can doubtlessly offset this value profit.

A superb instance of how vital it’s to run this process is to have a look at the next diagram, which reveals how this process eliminated 112 TB of storage.

Rewriting manifest recordsdata

Apache Iceberg makes use of metadata in its manifest listing and manifest recordsdata to hurry up question planning and to prune pointless knowledge recordsdata. Manifests within the metadata tree are routinely compacted within the order that they’re added, which makes queries sooner when the write sample aligns with learn filters.

If a desk’s write sample doesn’t align with the question learn filter sample, metadata might be rewritten to re-group knowledge recordsdata into manifests utilizing rewriteManifests.

Whereas Cloudinary already had a compaction course of that optimized knowledge recordsdata, they seen that manifest recordsdata additionally required optimization. It turned out that in sure circumstances, Cloudinary reached over 300 manifest recordsdata—which had been small, typically beneath 8Mb in measurement—and attributable to late arriving knowledge, manifest recordsdata had been pointing to knowledge in several partitions. This prompted question planning to run for 12 seconds for every question.

Cloudinary initiated a separate scheduled strategy of rewriteManifests, and after it ran, the variety of manifest recordsdata was decreased to roughly 170 recordsdata and on account of extra alignment between manifests and question filters (primarily based on partitions), question planning was improved by 3 times to roughly 4 seconds.

Selecting the best question engine

As a part of Cloudinary exploration geared toward testing varied question engines, they initially outlined a number of key efficiency indicators (KPIs) to information their search, together with help for Apache Iceberg alongside integration with present knowledge sources akin to MySQL and Snowflake, the provision of an online interface for easy one-time queries, and price optimization. In keeping with these standards, they opted to guage varied options together with Trino on Amazon EMR, Athena, and Snowflake with Apache Iceberg help (at the moment it was accessible as a Personal Preview). This method allowed for the evaluation of every answer in opposition to outlined KPIs, facilitating a complete understanding of their capabilities and suitability for Cloudinary’s necessities.

Two of the extra quantifiable KPIs that Cloudinary was planning to guage had been value and efficiency. Cloudinary realized early within the course of that completely different queries and utilization sorts can doubtlessly profit from completely different runtime engines. They determined to concentrate on 4 runtime engines.

Engine	Particulars
Snowflake native	XL knowledge warehouse on high of knowledge saved inside Snowflake
Snowflake with Apache Iceberg help	XL knowledge warehouse on high of knowledge saved in S3 in Apache Iceberg tables
Athena	On-demand mode
Amazon EMR Trino	Opensource Trino on high of eight nodes (m6g.12xl) cluster

The check included 4 forms of queries that symbolize completely different manufacturing workloads that Cloudinary is operating. They’re ordered by measurement and complexity from the best one to probably the most heavy and complicated.

Question	Description	Knowledge scanned	Returned outcomes set
Q1	Multi-day aggregation on a single tenant	Single digit GBs	<10 rows
Q2	Single-day aggregation by tenant throughout a number of tenant	Dozens of GBs	100 thousand rows
Q3	Multi-day aggregation throughout a number of tenants	Lots of of GBs	<10 rows
This fall	Heavy sequence of aggregations and transformations on a multi-tenant dataset to derive entry metrics	Single digit TBs	>1 billion rows

The next graphs present the associated fee and efficiency of the 4 engines throughout the completely different queries. To keep away from chart scaling points, all prices and question durations had been normalized primarily based on Trino operating on Amazon EMR. Cloudinary thought-about Question 4 to be much less appropriate for Athena as a result of it concerned processing and remodeling extraordinarily giant volumes of complicated knowledge.

Some necessary elements to contemplate are:

• Price for EMR operating Trino was derived primarily based on question period solely, with out contemplating cluster arrange, which on common launches in slightly below 5 minutes.
• Price for Snowflake (each choices) was derived primarily based on question period solely, with out contemplating chilly begin (greater than 10 seconds on common) and a Snowflake warehouse minimal cost of 1 minute.
• Price for Athena was primarily based on the quantity of knowledge scanned; Athena doesn’t require cluster arrange and the question queue time is lower than 1 second.
• All prices are primarily based on listing on-demand (OD) costs.
• Snowflake costs are primarily based on Customary version.

The above chart reveals that, from a value perspective, Amazon EMR operating Trino on high of Apache Iceberg tables was superior to different engines, in sure circumstances as much as ten instances cheaper. Nonetheless, Amazon EMR setup requires further experience and expertise in comparison with the no-code, no infrastructure administration provided by Snowflake and Athena.

When it comes to question period, it’s noticeable that there’s no clear engine of selection for every type of queries. In actual fact, Amazon EMR, which was probably the most cost-effective choice, was solely quickest in two out of the 4 question sorts. One other fascinating level is that Snowflake’s efficiency on high of Apache Iceberg is nearly on-par with knowledge saved inside Snowflake, which provides one other nice choice for querying their Apache Iceberg data-lake. The next desk reveals the associated fee and time for every question and product.

.	Amazon EMR Trino	Snowflake (XL)	Snowflake (XL) Iceberg	Athena
Query1	$0.01 5 seconds	$0.08 8 seconds	$0.07 8 seconds	$0.02 11 seconds
Query2	$0.12 107 seconds	$0.25 28 seconds	$0.35 39 seconds	$0.18 94 seconds
Query3	$0.17 147 seconds	$1.07 120 seconds	$1.88 211 seconds	$1.22 26 seconds
Query4	$6.43 1,237 seconds	$11.73 1,324 seconds	$12.71 1,430 seconds	N/A

Benchmarking conclusions

Whereas each answer presents its personal set of benefits and disadvantages—whether or not by way of pricing, scalability, optimizing for Apache Iceberg, or the distinction between open supply versus closed supply—the sweetness lies in not being constrained to a single selection. Embracing Apache Iceberg frees you from relying solely on a single answer. In sure eventualities the place queries have to be run ceaselessly whereas scanning as much as a whole bunch of gigabytes of knowledge with an purpose to evade warm-up intervals and preserve prices down, Athena emerged as the only option. Conversely, when tackling hefty aggregations that demanded vital reminiscence allocation whereas being aware of value, the choice leaned in direction of utilizing Trino on Amazon EMR. Amazon EMR was considerably extra value environment friendly when operating longer queries, as a result of boot time value may very well be discarded. Snowflake stood out as an excellent choice when queries may very well be joined with different tables already residing inside Snowflake. This flexibility allowed harnessing the strengths of every service, strategically making use of them to go well with the particular wants of assorted duties with out being confined to a singular answer.

In essence, the true energy lies within the capacity to tailor options to various necessities, utilizing the strengths of various environments to optimize efficiency, value, and effectivity.

Conclusion

Knowledge lakes constructed on Amazon S3 and analytics providers akin to Amazon EMR and Amazon Athena, together with the open supply Apache Iceberg framework, present a scalable, cost-effective basis for contemporary knowledge architectures. It allows organizations to rapidly assemble sturdy, high-performance knowledge lakes that help ACID transactions and analytics workloads. This mixture is probably the most refined method to have an enterprise-grade open knowledge atmosphere. The supply of managed providers and open supply software program helps corporations to implement knowledge lakes that meet their wants.

Since constructing a knowledge lake answer on high of Apache Iceberg, Cloudinary has seen main enhancements. The information lake infrastructure allows Cloudinary to increase their knowledge retention by six instances whereas reducing the price of storage by over 25 %. Moreover, question prices dropped by greater than 25–40 % because of the environment friendly querying capabilities of Apache Iceberg and the question optimizations offered within the Athena model 3, which is now primarily based on Trino as its engine. The power to retain knowledge for longer in addition to offering it to numerous stakeholders whereas decreasing value is a key element in permitting Cloudinary to be extra knowledge pushed of their operation and decision-making processes.

Utilizing a transactional knowledge lake structure that makes use of Amazon S3, Apache Iceberg, and AWS Analytics providers can enormously improve a corporation’s knowledge infrastructure. This enables for stylish analytics and machine studying, fueling innovation whereas protecting prices down and permitting the usage of a plethora of instruments and providers with out limits.

---

In regards to the Authors

Yonatan Dolan is a Principal Analytics Specialist at Amazon Net Providers. He’s situated in Israel and helps prospects harness AWS analytical providers to leverage knowledge, acquire insights, and derive worth. Yonatan is an Apache Iceberg evangelist.

Amit Gilad is a Senior Knowledge Engineer on the Knowledge Infrastructure staff at Cloudinar. He’s at present main the strategic transition from conventional knowledge warehouses to a contemporary knowledge lakehouse structure, using Apache Iceberg to reinforce scalability and suppleness.

Alex Dickman is a Employees Knowledge Engineer on the Knowledge Infrastructure staff at Cloudinary. He focuses on partaking with varied inner groups to consolidate the staff’s knowledge infrastructure and create new alternatives for knowledge functions, guaranteeing sturdy and scalable knowledge options for Cloudinary’s various necessities.

Itay Takersman is a Senior Knowledge Engineer at Cloudinary knowledge infrastructure staff. Targeted on constructing resilient knowledge flows and aggregation pipelines to help Cloudinary’s knowledge necessities.