Delta Lake Delta Rust, 19 Jul 2022

Previous Meeting Next Meeting

⏯

youtube image

►

From YouTube: Streaming Data into Delta Lake with Rust and Kafka

Description

Scribd's data architecture was originally batch-oriented, but in the last couple years, we introduced streaming data ingestion to provide near-real-time ad hoc query capability, mitigate the need for more batch processing tasks, and set the foundation for building real-time data applications.

Kafka and Delta Lake are the two key components of our streaming ingestion pipeline. Various applications and services write messages to Kafka as events are happening. We were tasked with getting these messages into Delta Lake quickly and efficiently.

Our first solution was to deploy Spark Structured Streaming jobs. This got us off the ground quickly, but had some downsides.

Since Delta Lake and the Delta transaction protocol are open source, we kicked off a project to implement our own Rust ingestion daemon. We were confident we could deliver a Rust implementation since our ingestion jobs are append only. Rust offers high performance with a focus on code safety and modern syntax.

In this talk I will describe Scribd's unique approach to ingesting messages from Kafka topics into Delta Lake tables. I will describe the architecture, deployment model, and performance of our solution, which leverages the kafka-delta-ingest Rust daemon and the delta-rs crate hosted in auto-scaling ECS services. I will discuss foundational design aspects for achieving data integrity such as distributed locking with DynamoDb to overcome S3's lack of "PutIfAbsent" semantics, and avoiding duplicates or data loss when multiple concurrent tasks are handling the same stream. I'll highlight the reliability and performance characteristics we've observed so far. I'll also describe the Terraform deployment model we use to deliver our 70-and-growing production ingestion streams into AWS.

Connect with us:
Website: https://databricks.com
Facebook: https://www.facebook.com/databricksinc
Twitter: https://twitter.com/databricks
LinkedIn: https://www.linkedin.com/company/data...
Instagram: https://www.instagram.com/databricksinc/

A

Welcome to streaming data into Delta lake with Rustin Kafka, or if you prefer, how Scribd reduced data ingestion costs by 95 percent um so I'm. My name is Christian Williams I'm, a vena software engineer since 2006.

A

um I'm, a contributor to Kafka, Delta, ingest and Delta RS, which are to the breast projects that we'll be talking about today. I work for Scribd our mission is to change the way the world reads. We are the initial sponsors of the Delta RS and Kafka Delta ingest projects I've been at Scribd since 2019 and I can tell you.

A

It is a great place to work um so first I'll go through the agenda for today's session I'm, going to start by giving some historical context leading up leading up to the Rustin implementations that we'll be discussing today.

A

um I'll just describe the problem, we're trying to solve with rust and go into some details about our rust implementations, I'll mention the performance. We've observed the architecture and design and what it looks like in our production deployment.

A

uh Finally, I'll finish up by talking about the future of Kafka Delta ingest and just sharing some general thoughts and experience experiences that we've we've had from rust. At the end, we'll do some q a.

A

All right so we're going to start with a bit of History.

A

So before we did streaming at Scribd, we had self-managed Hadoop and Kafka clusters. Kafka wasn't the only platform we used for event, logging um and even even where we were using it. The the Kafka topics we had had a variety of unstructured data formats, some Json, some CSV um and just some kind of some bespoke custom encodings um and we didn't have any streaming consumers we had. We had uh Kafka producers, a bunch of applications, logging events into Kafka, but we didn't consume, consume those in real time anywhere.

A

um Instead, what we had was in our nightly pipeline, we would read a day's worth of Kafka messages and load those into Hive partitions in our Hadoop cluster. So we wanted to do a few things we wanted to to standardize on Kafka for event, logging.

A

um We wanted to standardize on Json for our top topic data format and we wanted to convert those and ask those those nightly tasks that loaded data into our Hive catalog into streaming jobs. We wanted to do that to take some pressure off of our nightly run time and also to make event data queryable in real time.

A

So the first step we took was to migrate to a managed Kafka. We chose AWS msk managed streaming for Kafka for that and we also at the same time, we began adopting databricks and Delta Laker um and some some of the things we liked about Delta, we loved the optimize and vacuum commands, because that makes it easy to clean up and compact small files.

A

Streaming workloads have a tendency to produce a lot of small files, um and we like the fact that you can use Delta tables as streaming sources as well, so we could have streaming pipelines that start with Kafka and then continue on with with Delta tables.

A

So our first approach for for writing events to Delta Lake was a generic spark structured streaming application.

A

um So we we built this application that could be used generically using some config parameters, and it was a pretty simple spark query like the one you see on the screen here to just read: Json from Kafka write it to a Delta table. We used the the Delta table, schema to define the Json deserialization format and, in general, the table schema for each event.

A

Stream was the same as the as the Json structure, but with a couple simple transformations to include a date partition and also add Kafka metadata Fields, so that we could keep track of each record uh exactly what partition and offset and topic it came from uh when we queried our Delta table.

A

So we had a lot of topics to write to Delta and we had demand for a lot more. So our and also our topic volume, our topic volume was was highly variable. um So we had some very large topics and some very small topics we didn't want to have a databricks job per stream running this spark application. So we lumped some streams together into coarser grain databricks jobs.

A

That would run many queries simultaneously, each one taking a particular topic into a particular table, some some downsides- and that's what you see here so like on the Kafka side. You have like the purple cylinder. uh You know three of those. You know we have two topics going into that that are being consumed from one spark job and within that one's work job we have two different streaming queries running doing the work, so this was especially our strategy for for small volume.

A

Topics for large topics we're a cluster using a dedicated spark job didn't didn't, seem too too wasteful. We would give it its own dedicated job, so some downsides of this approach, uh one was so our streaming queries weren't, always isolated from each other.

A

So if something went wrong in one like the whole job fails, not just that one query and you know we tried to group topics together logically, but this was a challenge, because there wasn't a really great grouping mechanism that that was kind of a a good reason to put topics into the same job.

A

uh So, aside from this setup, we also created a nightly task for running optimize and vacuum on each of our streaming tables.

A

After uh so uh after after we implemented this, we we quickly realized. We wanted something better. um Some of the some of the other downsides was uh the cost, was high for running all of these spark jobs. 24 days 24 hours a day, seven days a week, some of our jobs would fail and restart occasionally and- and there wasn't always a clear cause when looking through the logs um the operational plane for managing these. These jobs was a lot different than we were used to for a lot of our other services.

A

We typically have like a big datadog dashboard, with lots of metrics and and logs and um with our streaming jobs. We didn't have that um at least at least not as many metrics as we're used to also we couldn't affect. We couldn't effectively Auto scale, um so we started wondering. Could we append to Delta tables without Spark?

A

Could we do it with a bunch of little headless fargate containers that run like traditional services and aren't part of a spark cluster.

A

So this is this is where Delta RS comes in, so my former colleague QP um I'm told he went Insanity Wolf and did a rust bike which implemented the Delta transaction protocol and a storage abstraction, um and he called this Delta RS and he created an open source repo for it. So at this point we had a rust implementation of the Delta protocol in hand, and then we were like all right. Well now, all we have to do is Implement streaming. Rights sounds easy right.

A

um It kind of does sound easy because you know this is a very specific problem that we were trying to solve. We only needed to append records to these ingestion tables. We didn't need to do any filtering, joining aggregating or sorting. We basically didn't need any anything that would require a spark Shuffle or coordination or detailed coordination between nodes in a cluster.

A

um We only so we only needed this tiny subset of features we only needed to append um and we figured we decided. We could use kafka's consumer group feature for partition assignment so that we could avoid service to service communication, but um between uh tasks handling the same stream.

A

um Also we could we figured. We could continue to rely on our nightly Delta, optimize and vacuum to clean, clean up. In case we produced small files. While we were streaming.

A

So, that's why it's easy. It is still kind of hard, though, first and foremost, we have to be very careful to not corrupt the Delta transaction log. We have hundreds of batch jobs Downstream from our ingestion tables, and we don't don't want to break any of those Downstream jobs because they're they're hard to fix and they take a long time to get running again.

A

um We knew we had at least a few topics that were large enough, that they would require multiple writers.

A

And we also had the problem that we use AWS and S3 and S3 doesn't support, put if absent semantics, which makes it really hard to well. It complicates at least the keeping the transaction log pristine and not corrupting it I'm not going to talk a lot about the missing put if absent semantics, if you're not familiar with with that concept, QP actually gave a great talk yesterday that goes in depth on on how exactly how we solved um result for that in Delta RS.

A

um So the other thing is, we don't want a bunch of uh duplicates or lost data, so we got to be really careful with our Kafka offset tracking and rebalance handling.

A

So we said you know given given the the easy parts and the hard parts, we still feel pretty confident we're going to do this in Rust.

A

So just the basic setup is we have Delta RS, providing the Delta transaction protocol and the storage abstraction and within Delta RS we use the the S3 storage abstraction, which includes dynamodb, in order to enable um you know: Atomic rename of transaction log files to kind of protect us from from corrupting the transaction log. And on top of that, we added Kafka Delta ingest to consume from Kafka buffer messages, and you know, write to Delta Lake periodically.

A

So Kafka Delta ingest is an it's basically, an implementation of that same spark streaming. Json query that I showed you earlier uh a simplistic example of starting. It up looks like this invoke: Kafka Delta ingest. You tell it to ingest from in this example my topic, and you tell it to write to a specific location, and in this case it's writing to a Delta table stored under my bucket my database, my table.

A

Oops, here's a slightly more realistic example: there's tons of command line Arguments for Kafka Delta ingest to kind of tune, the demon to run uh according to this, the size and and uh the size of the topic that you're handling and how? How often you want your your rights flushed um and how how large you'd, like your your files to be um also, you know, there's an argument mentioned here uh that lets Kafka Delta ingest know to write transaction log checkpoints um after every 10th commit now.

A

We also have some basic support for message Transformations, and this is how we derive our our partition column as well as ADD Kafka metadata to the message before it's written to Delta.

A

um So let's talk about performance a little bit, so we don't have any official benchmarks, but we have plenty of prod examples because we're running somewhere between 70 and 90 of these Kafka Delta ingest services in production. Right now, so I've selected three examples to kind of go over. You know what kind of performance we're seeing for different workloads.

A

None of these are giant, but we have a decent sized one. In example, three uh we have, you know, like I, said a large variety of low volume topics and a few high volume topics, so example one we we're running a single fargate task with one vcpu, this one's slightly over provisioned, because it's it's only using it's using less than 40 percent of the CPU. It's handling about 3000 messages, a second, the average message size is about 2, 000 uh or I'm.

A

Sorry, 2K uh example: two we have a single fargate task running on a quarter vcpu, which is really cheap handling about a thousand messages per second message. Size is about 1.2 k, third example. This is one of our larger streams. We have three tasks: we're doing a 2v, CPU fire gate on that one we've got about 8 000 messages per second, but that one varies a little bit we climb over 10 or 11. Sometimes the average message size on that is 3K.

A

So that's kind of this. What the Kafka Delta ingest deployments look like for those three examples: here's here's a little bit of a cost comparison, comparing Kafka Delta ingest to our prior spark structured streaming approach, which I didn't mention this before. We called that one Kafka Delta streams that was not open source by the way that was just an internal application.

A

So, rather than going over all the instance types since we're already getting a little short on time here, I'm just going to mention the the pricing difference, differences that we realized. So we had, for example, one that pipeline was costing us about 1500 5200 annually with Kafka Delta ingest. It's costing us about 400 example: two we went from thirty nine hundred dollars annually to 100 and in example, three we went from about 23 Grand annually to 2200.

A

So why is our rust implementation so much more efficient? So first off spark is a big massively powerful distributed processing Engine with tons of features, uh and you know we don't need all that. So uh with a specific use case, we're solving, for here was a pinned. Only Delta workloads, they're that are pretty simple. We've got a you know, read some data from Kafka and occasionally create a new file and update the transaction log.

A

So next rust is way more efficient than than Java and uh also finally, fargate is cheap and a quarter vcpu is is actually plenty for a lot of our streams.

A

We'll go over the architecture and design a little bit here all right. So um the way to look at this diagram is, uh you know it's kind of an infrastructure diagram with numbered edges, so I can kind of talk through each Edge and describe what's happening. Data flow wise, there's also an exploded view here, which I'll get to so. First, the Big Green number one Circle producers send messages to Kafka. We have lots of different producers in Britain and various languages um step. Two.

A

We have our ECS tasks read from Kafka, and here we have the exploded view, which shows you know. Depending on the the topic volume, we might have more than one task handling that topic or if it's a small, a small topic we we might have a single task handling it.

A

um Our fargate sizing is configured based on the topic. um Each task runs a Kafka, Delta, ingest container and a data datadoxide car.

A

So uh the way we run Kafka we have read and write ACLS defined for all of our topic, so that only specific applications can read or write. So we set additional Kafka properties in order to do that authorization.

A

So, on step, three eventually Kafka Delta ingest will write a parquet file and a temporary Delta log entry to S3 step. Four four and five are both implemented as part of the Delta RS commit so first step. Four, we acquire a dynamodb lock and step five. We do a rename of our temporary transaction log file to the actual um specific version file name required to represent a valid Delta log transaction.

A

Finally, with uh yeah. Actually that was number five, um so outside of the primary data flow, we also have our nightly optimized and vacuum jobs running in data bricks, which also use our dynamodb lock prior to running the optimize command.

A

So some of the key aspects of the multi-rider design, so that we can have multiple tasks, handling the same Kafka topic. One Delta RS includes a dynamodb lock to fill in for missing S3 put if absent and which I mentioned, but I I wanted to mention it one more time here so I could give you the link to qp's talk if you're interested highly recommend checking that out.

A

It goes really deep on that, um and another key aspect is that the partition offsets that represent the data accumulated from a specific Kafka partition are written into the Delta log.

A

um That way, you know if a container goes down another one comes back up and picks up that partition. It can read from the Delta log and see okay. This is where we left off. I need to start from that Offset, you know, do a seek and start consuming again um and uh the the last key aspect is as Kafka Delta ingest is buffering data. If a consumer rebalance happens, we take a very simplistic approach and we we just drop the buffer data re-seek to our initial offsets and reconsume.

A

All right so now, I'll just I'll describe what this looks like for us in production. So we have a multi-branch Jenkins pipeline that does a Docker image build that builds. It basically builds a Docker image.

A

Of Kafka Delta ingest uploads it to our AWS ECR registry.

A

We also have a terraform pipeline, which includes a module that wraps all of our ECS config and our Kafka Delta ingest parameters for each topic and generates a task definition for each service and that definition comes from a config map for each topic.

A

um We provision datadog monitors as well for lag and critical errors and our task definition supplies our AWS Secrets manager aren't that are required for our Kafka topic. Authorization.

A

um Here's a little snippet of what our what our terraform config looks like um not going to go into detail since we're running out of time here. But uh you know it's a pretty simple way to config to configure one to configure a Kafka Delta in just service to yeah um Implement. One of these streams.

A

um And of course, we have datadog dashboards that we can filter by environment and table, and you know see all the wonderful metrics that we emit to our datadog Sidecar.

A

So just a few short words about what we have in mind for the future. um We have a few Auto ongoing challenges that that are kind of top priority. One one our Auto scalability needs Improvement Kafka rebalance, makes this hard I, currently have a branch I'm working on to try to improve our our rebalance handling to make Auto auto scaling more effective.

A

There are some memory improvements we need to make in Delta RS, because we're storing the transaction log in row base format right now, and we need to store that in a column based format to reduce some memory pressure pressure issues.

A

You know the other two bullets I'm going to skip and then some general enhancements Beyond. These aren't causing issues for us at this time. But so some things we'd like to see to see happen for coffee, Delta ingest in the future, is additional. Cloud support uh trigger wants semantics because for these low volume topics I'd love to do that thing.

A

You can do in spark where you can do like a trigger once a day type thing and just read everything: everything for up to the current offset in Kafka and write that uh and then finally additional data data format, since we only handle um Json so far, so we love the value we're getting from rest at Scribd, we've got other stuff written in Rust as well.

A

We have a syslog endpoint that produces to Kafka I've, seen a couple of awesome rust topics uh at the um at the summit this week, one of which is about data Fusion, which I recommend you checking out. The other is qp's talk where he was talking about state right for formal verification of distributed problems, um and that's it we're ready for Q a as soon as I give some some thanks. uh Misha was one of the primary contributors on Kafka Delta, ingest, Tyler and QP uh really helped get Delta. Rs and coffee.

A

Delta ingest kicked off Neville and Florian, where two other folks who were early contributors and also um I'd like to extend an invite to anyone interested to come work with us at Scribd. It's a great place to work like I said: I've been there for a while and uh I love it. So that's it and I think we're ready for a q, a yeah.

B

All right everyone, let's thank Christian for his wonderful presentation, uh questions just raise your hand, will come by with the microphone. I saw a hand, go up in the back first and then I'll come up here.

B

We only have a couple minutes left uh before lunch, so maybe just these three questions, and then uh people can talk to the speaker in the hall.

C

uh Well, thanks for the talk, it was interesting to hear uh you mentioned uh datadoc a couple of times. So can you go into that topic more like? What do you specifically choose to log into datadog and uh what's the reason behind this decision.

A

uh I heard you could you uh clarify part of that I heard you say datadog but I didn't hear the specific. Oh.

C

Yeah yeah I was I was saying that you mentioned datadog a couple of times. So what do you uh did you choose to log specifically in today, the dog, and why.

A

Behind this decision.

C

A

So we've got uh a kind of a laundry list of counters that we log, so um we log every every time we um deserialize a message. uh Every time we transform a message, not all messages get transformed and there you know. If a failure happens, we also log um the the transform failed. uh We log the every we periodically log the number of Arrow record batches that we have stored in memory.

A

When we complete a write, we log the size of the parquet data file that we wrote. um Those are the big ones that I can think of off the top of my head. uh Oh we periodically log the current lag that we have from Kafka, so we'll peek at the current offset we'll look at the last offset. We have buffered and we'll submit a metric for for that to data dog as well, so that we can use that for data.monitors and alert if our lag gets too high.

B

Well, that's all the time we actually have today uh again thanks to Christian uh for giving this wonderful presentation today, and you can probably find them out in the hall for additional questions. Otherwise, it's time for lunch enjoy.