Beyond Kafka and S3: Python Data Pipelines with HTTP-Native Bytestreams

Speaker 1 [00:02]

Good afternoon everyone. Welcome to the session Beyond Kafka in S3 Python Data Pipelines with HTTP native by streams by test streams with the Johannes Drudge. Hope you said correct.

Speaker 2 [00:21]

Okay. So, yeah, I'm Johannes. I'm a data engineer, and today I'm going to show you a different perspective on cross-organizational data sharing with Python and in general, a minimalistic way to look at it. it. Who knows that? Most people. Do you still use that a lot? Yeah? I just looked it up. It's over 50 years old, actually. It's the pipe. It's the shell pipe that was introduced in 1983. You may call me old school, but when I look at this, I see beauty and purity. It's really the essence of what data pipelines are. It's a park. It doesn't care about what flows through it. It's just bytes. It's minimal from a user perspective, so you don't need to add anything than just the pipe sign in the middle, and it's composable. We all know that you can build very complex programs, shell scripts with pipes, and process commands basically. So this is going to be the baseline for our talk about cross-organisational data sharing. When we talk about cross-organisational data sharing, I brought two use cases, pretty arbitrary, but they are supposed to highlight the differences between organisational data sharing and data sharing between systems and programmes and so on. One use case is a traditional one that you have when you have suppliers and buyers. Usually you need to update stock information from the supplier, and the buyer might have a pricing pipeline that needs to be updated. Simple case. The second one is slightly different. It's when hospitals run clinical trials. There is a contract research organization that runs the trials. They are basically paid by pharma, and they need to get access to patient information in an anonymized way. So there's also a need to basically exchange information here. So in both cases, these two parties have nothing in common, almost, so they need to share something. They need to share infrastructure, they need to share trust, they need to share protocol data formats, basically, they need to negotiate, they have meetings, they have an onboarding project, and what we can see is that the harder the decoupling of the organisation is, the higher is the cost of the solution that you have to build. So the cost scales with the decoupling, and it's not a one-time cost, it's something that's ongoing, because you have the operational overhead, basically. So when you look at it from a historic perspective, let's start at the function level. When you program, you are also passing data, and when you call a function, it has a signature, so you're basically passing data. But the beta data is in memory. You probably just pass a pointer to a space, a memory section, and That is very fast, there is no alignment cost basically, it is all built in. When you move to inter-process communication, and the pipe is a version of IPC, then two processes on the same machine communicate. So you need to pass the information from one to the other. So basically the data moves from the memory area of one process to the other one usually. So then the internet came, and also local networking, and you add complexity to this. So this is where we enter the realm of remote procedure calls, FTP, SSH, HTTP. So they rebuilt the whole thing, and they added basically transport security and reliability as the main features. So when we move on now, then the question is, what's different with organizations? Is it just network, or is there something else? So just imagine two companies, like maybe the hospital or a medical company. The other one is a cloud-native one. So they run a different infrastructure. They have a different technology stack, they do deployments, the other one does updates, they have different compliance regimes, they use different technologies for data interfaces. So there is a huge alignment cost here. So they have to build a solution that works for both of them, and that's usually what happens. They find the lowest common denominator, which is usually copying CSV files, if you remember maybe. But there is also an alignment cost when it comes to the organisation. So each company works differently, and you have to have meeting calls and so on. So the question is can we make it any better for organisations, what we have? And the perspective that I'm showing here now is that composability is a good design response for this challenge because it helps to separate the layers so that we can treat them separately. So usually when we're talking about the data transport, which is different from the serialisation format and the data quality and data structure, then the transport protocol should actually govern the transport itself. I put data on a USB stick, and I bring it somewhere, and then I copy it over. So that's the transport. The access is basically I give it to someone I trust. And everything else is just call or cite, because that enables a multitude of different applications that are more specific to these parties that try to build solutions. So the question is, how does the protocol for this strongly decoupled case actually look like? So what would be the ideal protocol for it? So when we look at it from what we have, then it's getting complicated. It's getting complicated because we have different tools made for different things. So I put them in three dimensions. One is the infrastructure coupling, it's basically saying how much infrastructure we need in the middle, or on one of both sides. The other one is the technology coupling, it's a little bit fuzzy, I know, but this is like the spillover, how much do we impose on our partner? And the coordination model is about pub-sub or event-driven communication request-response or push or pull, basically. It's different ways how to describe it. So when we look at tools, and we're looking at Kafka and S3 and web APIs as representatives, you can drop in any tool that you know, like take RabbitMQ instead of Kafka, or you can use SharePoint or SFTP instead of S3. So the general things that we observe is that when you're looking at the message brokers, they usually have a specific protocol that's kind of spilled over in the technology. It's the second part. There's some kind of a cluster broker in the middle that needs to be co-owned or at least shared from one side. It's a huge infrastructure part, and the coordination model is just push, so you push it in on the one side, it drops out on the other side, so you can't really request data. For S3, the infrastructure coupling is actually pretty good, because you just set the permissions, and somebody of course has to rent the bucket or host a minimal server, but that's quite okay. from a technology side, it's also okay because it's based on HTTP, and that's a very accepted protocol. When you look at the coordination, it's basically the same. It doesn't have coordinated coordination, but usually what we do is we trigger, we just pull what we see, and then we can emulate something like a PubSub model, but it's still just one way. You push, and receive. When you look at the REST APIs, things are a little bit more complicated. Infrastructure is asymmetric as opposed to the other ones, so one side has to be the server, and that comes with security, with development, with data model, maintenance, and so on. When we look at the technology, it's okay, but REST APIs have one problem. They usually cross the layer between the transport and the data model. When you consider HTTP as the raw transport, that's fine, but that's not what a REST API looks like. It's really blending the endpoints on top, and that makes it not a clear separation. The nice thing about the coordination is it works both ways, but you have to design it when you make it. So either the one side is the server, the other one is the server, or you need to add webhook communication to it, so it gets complex. So, what would a good requirement look like to reduce the infrastructure technology and coordination coupling? Coordination coupling, no. Technology coupling and coordination flexibility, let's say. And that's why we go back to our initial motivation. When we look at pipes, and specifically named pipes, they were introduced 15 years later, actually. We see that we have a model that actually works quite nicely. So for those of you that I don't know name pipes on a Unix that well, it's basically a pipe but it adds a file system representative object and that inherits all the permissions basically that you have on the file system level. So it makes things work more standard, like with files. And what we see now is that it has very low infrastructure coupling, it's basically just pipe object provided by the kernel. It's a lean shared infrastructure and both sides are clients that we already saw. It just writes files. The technology has a very simple file-like interface so it works for any bytes, no format restrictions and the trigger direction is dynamic in a way that either the consumer can conduct first and wait for the producer or the other way around. And you can actually wait for the other one and trigger something. But that's not obvious at the moment. So the question, the real question is, can we make this model work across organizational boundaries? And in this case I would say over HTTP because there are really good reasons to use that protocol as seen for S3. That's why we introduced the ZebraStream protocol. It's really minimalistic. It's just basically a port of the pipe model to the cloud. As we see, when we model it for HTTP, then both sides are HTTP clients. The technology coupling is minimal as for S3, and the coordination model, as we will see, is runtime dependent. So it's really, really, really flexible in the way to build event-driven pipelines that cross organisations. So how does it work in detail? So this is the relay, which we also called the Data API, and it's really, really simple. It's just connects to requests. One is the put request, one is the get request. So there's an upload and a download that's connected in real time. So there's one channel per transfer, and it's scoped, it's exclusive. It's like a name pipe. It has an address. It's secured via standard transport encryption, has very specific access via scope tokens. That's basically it. It works through firewalls, it works through organisations because it acts like a web browser or any HTTP client. So when we send data through this, we must also define how this data looks like, right? So what does any byte mean? So for that, I'm just recalling, because when we think of streaming, we often think of message brokers and message broker system and event streaming. So what's the actual difference between a byte stream and a message? So messages are basically streams of batches. And each batch is size-restricted, it's discreet, bounded. It adds a little bit of routing information because these message brokers are made for multiple producers, multiple consumers. So it's a different use case. It's when we have a microservice landscape and want to, I don't know, send information to multiple microservices or these kind of things. So that is a strong restriction if you want to be very generic. So the size of messages is usually below one megabyte. If we increase it, then it's up to 100 megabytes. I think RabbitMQ allows up to 512 megabytes, but that's really pushing the limits. It's not made for it. So in general, they are protocol agnostic, so you can serialize anything. But in practice, you can't really put a video file in there or any file format that you know that's really awkward. So they usually use protobuf, JSON, Avro, these kind of things made for messages. So whereas the byte stream is what we saw with a pipe, it's basically anything. So it's a park. It can be a file. It can be an endless stream. It can carry Parquet, arrow formats, JSON lines, you call it. So how does that work on the HTTP level then? Now we move into Python. So this is really the standard way how you do requests in web requests in Python. When you do an upload, basically, you call request put, give an endpoint, and in this case it accepts also a generator, which means that you don't upload a file, but you upload a stream, an endless stream, which defaults to HTTP junked encoding. That's the way how you can basically transfer streams in HTTP. There is also the download part. It works the same way. And that's it. That's really, really simple. So the job is actually just for the relay to connect the upload to the download in real time. But what if, for instance, the consumer isn't really ready to accept data if we do that. So that's the timing problem. So when we go back to the name pipe, when a program wants to open a file for reading or writing, it calls the open call. At the open call, basically it returns, and then you can write or read. So when you do that on a named pipe, the open pipe open call blocks if there is nobody on the other side. So it basically stalls. With HTTP, that's more difficult, because once you build the upload or download stream, there are two ways, basically, to fail. And that is... The failure one is we We are just listening, we are trying to download, but the relay doesn't have any data because there is no producer. So we can't tell whether we are actually waiting because the producer is just slow, maybe it's a process that sends an event every one hour, or if there is really no such producer. So the signal is the same. And similarly, when we produce, we are writing to the relay, and the relay just blocks, and we don't know whether this consumer is just slow or waiting for something, it's blocking internally, but it's there, or if there's no consumer at all. So the signal is, again, the same. When we use pipes, like the shell pipe, there's a mechanism called back pressure. That means, of course, here, the producer stops writing because the relay says there's no buffer. You can't write the data, it just stalls. But the problem with HTTP is that the data is not just on the one machine, it's not in the cache of a single pipe provided by the kernel. It's distributed across the client side cache in your library, basically. The routers on the way. Maybe a reverse proxy in between. And then, of course, the data piles up and at some point it blocks because of back pressure. But the infrastructure that you're using is not so nice. They don't like data hanging around, so they just kill it at some point. So basically we need a different way how to coordinate. And that's why we need an explicit handshake. And the handshake is basically just a different API, we build a modular, so that we have such a flow. It's basically the first client, producer or consumer, doesn't matter, connects to the connect API and hangs waiting. This is a long poll mechanism. When the second client connects, both meet, it's a rendezvous system, get dispatched and sent to the data API. That's the relay that we just saw. We know both are there. They know we can transfer data. They connect to the streaming. They stream via the data API and transfer the data. Transfer completes. We are done. As we see, whether we do push or pull is not really an architecture decision anymore. It's a runtime decision. whoever waits first is the server, basically. And we have this explicit transition, which is also nice, because then we can also build in load balancing, for instance, at the application level, because we can send the clients to different maybe on-prem relays. So now I want to show you this in action, because I think it speaks for itself when you see it. So let's go back to the supplier-buyer inventory use case. So now I'm going to show you this with a CLI here, just because it's better to focus on the communication part here. Basically, All we need to align is the stream address, and it's the access token. So the access token is different for both sides, because this is a privileged separation. And we need the logic, and the logic here lives in the export-import. So suppose the supplier has a database or warehouse where the inventory data lives. Suppose the buyer has a pipeline which does the pricing. So the job of the export and import logic is basically the export extracts the data from the database or warehouse, sends it into the stream, into the channel, and on the buyer side we receive it and directly process it and load it into the pricing pipeline. The thing about this is that it's totally event-driven because we have this Connect API. The export import doesn't run like in a shell when we execute it, but it runs when the other peer arrives. So that makes it very easy to react to events in both directions. And the other observation is that there's no difference in the code. It's just a temporal variant, basically. So let me show this to you. I think this is it. Is it? No. This is it. So here we have the use case that I just saw you. I just put the import and export scripts here. They are really lightweight. to show you how simple this whole thing is. We just query the database, which is an SQLite database in this case. We serialise as JSON lines, just an arbitrary decision, that's our alignment basically at the format level. And we just close it then. And the import works the same. We read JSON lines, and then we are just logging something. There is no logic here, actually. We could save it somewhere. We could directly process it. It doesn't matter. I hope I have internet here. I just put the alignment stuff, the important stuff, into a config file. So we have the supplier. No. Let me see what we have. Inventory. So you can see it's very minimalistic. It just says, like, what's the stream address? It's like a global file system, actually. And then it has the token. So, we can run this with the config file format argument. And what was the name again? Inventory read. And when we say read, we have to do that as an operation. So the CLI is just wrapping the Python SDK, the Python interface, and it's part of the Python package. So we're just using it here for convenience. We could also write it all in Python. So if we do the same on this side, So this is now hanging on the Connect API. If we now do the same for app, so this has to be, I'm sorry. Now I'm getting, I'm being the consumer. consumer. And I forgot also to give the command that exports the data, because we didn't put it in the config. So hopefully on the other side, which is the consumer side, we do the same. We just replace the config file, say read, and pass it to the import inventory. Let's see if it works. Usually the demo moment. As you can see, so it took a little bit because the relay spins up in the micro VM when we request it. If we repeat this, then it's faster. So you see that it's not really made for the use case of a web server, so handing millions of requests and so on. But this is a different use case. We don't care about a connection latency of one second. So we can also do it the other way around. I don't know which one did we start first. I think this one, right? So now my terminal is a little bit broken. So I have to copy this, this web terminal here. So we just now, basically what we did before is we ran the supplier in waiting mode, in server mode, and the consumer just connected when it wanted the data. Now we'll do the other way around. the consumer is just waiting, and we push the data whenever we want. So this would be the PubSub model where you subscribe to a feed. So I think, is that correct? Let's see if it works. Okay. I think it worked. So, this just wanted to show you what it means in terms of usage simplification, but now I want to show you the real Python stuff, which is the file-like interface. We see that the concept maps to a pipe, so why not make it file-like. So the ZebraStream.io package, which implements the protocol, also implements it in the form of a file-like object, which corresponds to the IO base class, which means that when we open such a stream as a file, or a file-like object, we can pass it to any function that accepts files. So basically that means we can turn any code that expects to write local files instantly into a streaming application, a consumer or producer. And that's really nice, because this universal abstraction, it handles all this transport layer transparently then without code change. There's another nice thing about it, which I would call shared fate is when you have a workflow that exports data, then you copy data, copy it somewhere else, then you import the data. That's how data engineering works in many cases nowadays. So that means you persist the data on storage, and you don't... Basically, it leaves your scope for a little while. But when you do streaming, you can, you actually have a success return state, a return value, which basically is either worked or it failed. So now when we run the producer, save something as CSV, and pass a stream instead of a file, and on From the consumer side, we run Pandas read CSV with the read side of the stream, and we just get interrupted because something happened. It will propagate to the producer side with a meaningful exception message. And the other way around, if the producer is happily producing, and I don't know, there's a memory error or something, it crashes, the consumer will know because there's no end of file, it just gets interrupted, and it will also know that the data is just partial. So that makes it really easy to have a risk transition between the two parties when you transfer a file, for instance. Just for the transfer. You can also instantly process it, validate it, and if it doesn't validate well, you just make it fail. That's another possibility. So that's fighting the copy cascade problem. You can open in text mode, in binary mode, it doesn't matter. Whatever you want. For CSV files, we would use, as you can see, the text mode here. If you transfer something like Parquet files, which are not made for streaming, you can still do it, but you can maybe not process it stream-like. So you need to buffer it, save it to a temporary file, and then work with it. If you do something like Arrow IPC, which is made for streaming, or CSV actually works nice for streaming, then you can process right away. So it's the same interface that we know. And now we can also make it work for another use case that everybody knows, which is log streaming. So in this case, we're going to use the log package log guru because it works nicely. But we could also use the standard logging module in Python. So what we do now is we use a stream as a log sink and just have our Python program run and log something. And on the consumer side, now we're basically combining the Python level interface and the CLI, which makes it work on the terminal. And I'm going to show you that in another Jupyter notebook. So this is our LogGuru producer. As you saw here, we just opened this with a ZebraStream-specific open method, which returns the file-like object, and we call it stream, and we just add it to the locker. That's basically it. So two lines of code that you need to make this work. Let's just say that we have a different system which is supposed to receive the locks in real time, which I put here, which is log guru read. I put the stream address and the read token into a config again, and that is it. We just want it on the terminal. Sorry. My history is broken here. I just reinstalled this. I have to type it. What did I do? I forgot the read at the end. Now we go. So let's start this. So we import the package, right? We load the credentials from the environment. We open the stream as a file-like object. Here we do not do it with a context manager because it does not work well in notebooks. We add this to the stream, and let us see what happens when we lock arrow. That is the latency that we get when we send from this computer via the Wi-Fi to the relay running in Amsterdam, I think, coming back here. You see it is fast enough for the regular use case. So let's just repeat this. Okay? Works. So we can just close it. Looking at the time, I think I have to hurry up a little bit. So just one thing that I wanted to show you here. We can also have it at an outer flush. So we can say it should collect all the data and flush it every 30 seconds. Because we don't need real time for locked data. it's fine if it arrives within two minutes or five minutes or whatever you like. So this is a compromise between middle ground, between real time and batch, basically. So let me go back to the slides. So now we come to the real core. What did we do? Well, we decided to use HTTP because it is portable. It has issues, it has drawbacks, yes, but that is okay. We decided to use byte streams, any size, any format, and we made the relay stateless, which means both sides need to connect at the same time. There is no data buffering. And that's an informed decision, we want these features, these properties, because they give you what we just saw. The interesting part is actually the security model. So when we said we separate access and transport layer from the data structure and data format, that enables us to use end-to-end encryption. So that means when we send sensitive data, we don't have to trust the cloud provider, we don't have to trust the relay provider anymore. So if we do per chunk encryption, we can also do it in real time. So that means if we apply that to the log streaming, we can easily make it encrypted. And that's just by adding a single passphrase here. So this implements the, maybe some people know it, the AGA-H encryption, so you're not supposed to call it H, I think, which is an established file encryption algorithm, but adapted to a real-time case here. So this is built into the library. That's nice. Which brings us back to the second use case, the hospital. So when we put together all these pieces, we can say that in the hospital case, maybe we want to use the client pull mode, so the research organisation that wants to pull the data to compile a compliance report with live data. The hospital is the server, it waits with the client and does the database export on demand. Actually, there's a patient system, a patient record system that needs probably to anonymise the data so you can just build it into the pipeline. Then, if you add end-to-end encryption, you get the full picture. Patient data is only accessible to the receiver, the consumer, which is the research organisation. And the net result is basically the hospital doesn't have to open ports. It just works. No problem with infrastructure. No VPN. The CRO gets the compliance report when it needs it, and the relay just doesn't care. It doesn't see any data, it just provides the transport as a service, basically. And that's really simple. And opens a lot of use cases, I think. So this is basically the offer. It's an open protocol. Here's the GitHub project. Go, use it, play with it, criticise it, contribute if you want. The client side is fully open source. It's basically a process, a development process, so it's not complete yet, I would say. But it works, as you see. And it works for any size of data. You can stream a terabyte of data, probably, in theory. But maybe you don't want to do that. Maybe you want to chunk the data, because basically you know that if you download a file and then your train runs through a tunnel, it fails and then you have to repeat it. So that's one of the drawbacks. So in total, I would say the pipe properties are retained. This thing is opaque, it's minimal, it's composable, and the addition is it works across organisational boundaries because we're using a technology stack that's made for it. That's it.

Speaker 1 [41:06]

Thank you very much. We have some questions for you. Is there any way to self-host ZebraStream?

Speaker 2 [41:16]

Yes. As I said, the client side is fully open source. I have to say, the data relay is not at the moment, because it is a ping-pong game as we iterate through the process, but it is based on open source components. There is a project which has been around for ten years now, it is called Piping Server, which implements the logic of the relay. You can make that work either, or you just use this. It's currently in open beta, so it doesn't cost anything. Just do.

Speaker 1 [41:49]

Thank you. Another question is, is this suitable for big-slash-heavy data operations, and is it production-ready?

Speaker 2 [42:01]

So, I think we are lacking a few, to answer the production already, I would like to say it, but I cannot prove it. So, it's based on solid components, and it works up to gigabit speed, if you allow it to do that, gigabit links. It's not heavy on the relay, but it adds overhead. It has a double encryption for instance, it breaks the transport encryption in the middle and re-encodes it and so on, but that's all secondary for the use case that I showed you. So I would say the seed spot is in the middle, between what you cannot do with message brokers. If you want to use it for CERN research data in terabytes, we don't know. But in theory, it works.

Speaker 1 [42:50]

thank you very much and the last question is can the message streaming be integrated to be used with salary or the task framework of Django how good is it handling concurrency

Speaker 2 [43:07]

So yeah, it's all about integration. So I'm I'm I would say it can be integrated everywhere So the core is async. So we even result though we saw this file like interface, which is fully sync It's just basically a portal to the async core. So it it's it's it's built with HTTP a a IO. I think that's the name And it's it's well, it's fast. So it of course it can be integrated integrated. There's no special treatment of this HTTP request. It's just the same that you use in your regular HTTP programs. So yes, I would say.

Speaker 1 [43:50]

Okay, thank you. There's one more question. What type of open source contributions are you looking for, especially on the Python side?

Speaker 2 [44:00]

Very good question. So this is meant as a portable protocol, so we are looking for integrations into all frameworks. So they can be at the language level, of course. We don't have a good TypeScript repo, so we are Python first, because that's where the data people are, and this is about data. But you can use it to build decentralized web apps. There's a whole dimension in making it web-native, so you can deliver web applications with it. You can do crazy stuff, but for the Python side, I think we provide the reference implementation, which I think is a good baseline, but you can build patterns around it. You can build event streaming, you can build it into higher-level communication systems, and that's what we we are looking for right now for the Python case. For the rest, general integrations would be nice.

Speaker 1 [44:59]

Thank you very much. I see there's another question, but we started a bit over time due to the technical issues, so I would suggest that you can meet Johannes for a coffee and ask directly the question, have a nice talk over the topic. So thank you very much for your presentation and for your questions.

Speaker 2 [45:22]

Thank you.