SBTB 2015: Paul Kinsky, Scraping Reddit with Akka Streams
Recording: SBTB 2015: Paul Kinsky, Scraping Reddit with Akka Streams
right so this talk is about reactive streams will be going over the standard how it works to implement back pressure really quickly then we'll sort of dive into an example that uses akka streams of reactive streams implementation to scrape reddit chosen simply because there's no auth involved in doing that and yeah so hopefully by the end of this you'll have a working understanding of how to build a simple pipeline that uses some of the different active streams features all right let's dive in first I'm Paul Kinski man platform engineer at nitro web document productivity company we made PDF smarter if you want to find out more you've got a booth about over there come see us right so streams you've heard a lot about this various places to the conference you have real-time events streams do you want to publish real-time data if Big Data of course which is very easily viewed as a stream when you can't fit the entire data set in memory and of course you have streaming media right you have netflix you have spotify those are all famous so skip over that two kinds of streams essentially distributed the ones where you might create a single stream pipeline that crosses multiple machines without having to manage the transfer of data from machine a damn machine be and so on sparks storm when you're training models you're dealing with big data it's not really what this talk is about more local streams or you can say specify your streaming pipeline with a lot more detail and use it for really granular control so first case complex logic it's a sort of fictionalized example of something I'm working on at work but let's say you listen to Kafka you get image upload events for each event you pull the image from s3 you create a thumbnail upload the thumbnail to s3 and then publish an event saying that the thumbnail was created right you might not want to do that in spark because you might not be able to ensure data locality right when you grab the image from s3 you don't want to pass the full image from machine to machine or you might just want like say to deal with this like native interface for something anyway or second case granularity of control right there's a thing you can do with a cat HTTP that's built on top of a cos streams it's very useful you can take a single request as sort of a tuple of header and source of byte string where you don't actually get the body of the request until you look at the headers that would be great looking dealing with DDoS attacks for example if someone's trying to upload a gigabyte you don't need to get that gigabyte in memory before you can like actually look at the header and disallow it right and then of course is a building block for distributed systems any distributed system has to be built by stream pub lines that cross from multiple from system 80 system be there's no like magic that just happens there you actually have to build that at some point all right so let's look at how you actually do a stream right you're transmitting data from some publisher to some subscriber across the network between threads whatever point as data goes from publisher to subscriber there are two traditional ways to do that you have push you pull in push case the publisher initiates the transfer of data right so it just sends data to the subscriber as quickly as it becomes available problem and then the subscriber is forced to handle all data pushed by the publisher it can use a buffer so it can like sort of smooth over spikes but it has to just handle any data that's pushed of it or pull sort of the opposite where it's the subscriber that initiates the transfer data setting of the subscriber hitting the publisher with requests problem there then is publisher replies data only if it is data so you're just sending a lot of requests over the network with like that are completely wasted if you're doing pull mode and if there's no data at the publisher so just make that really clear right you have this two-by-two thing so the subscribers fast or you're doing push that's great the subscriber can handle whatever you push to it but if your producers faster you doing push the new either for buffer overflow you have lost data and we're dealing with machines right they have finite memory space so you can't just have a buffer that grows indefinitely so you're forced to either lose data or have just like a buffer overflow and crash in which case you lose data anyway and if you have pull so it's a little bit safer but if the subscribers faster you're just sending a lot of requests over the network they can answer it with nothing network congestion right there's a reason we don't do polling for everything and then of course we pull the producers faster it always has data you can always pull data from that's great so we really want to do you have two different scenarios you want to switch between push and pull based on the scenario ideally dynamically so how reactive streams does this it has a demand back channel so you send data downstream and you send demand upstream so how that works is the publisher can only send eight a downstream if demand exists if there's surplus to man to fulfill and it can send data in batches right if there's a thousand demand it can send a thousand at once it's not limited to sending one element for request or per message and then the subscriber will signal more demand the publisher will fulfill it etc so let's look at how that works with the subscriber fat is faster and publishes faster scenarios so subscribers faster publisher always a surplus domain to fulfill or surplus dammit surplus demand to fulfill and then elements are just pushed downstream as they become available right so just send stuff downstream there's no requirement for a message from the subscriber to do that as essentially push model then let's say the publisher becomes fast alright so then there's usually no surplus demand then demand from downstream is responded to with data that's essentially the pull model because you're sending a message upstream sending like yes I have demand for data and that is responded to more or less not in a request response cycle but it directly causes a message with data or more messages with database to be sent down strand so you have two flows here right you have demanded data so if you fan out one stream to to that means the demand is merged and the data split and vice versa if you merge the data is merged but the demand is split right so that's reactive streams the idea is that it's a standard it's been worked on by engineers from netflix twitter type-safe of course Red Hat various companies so hopefully as these companies open-source products or frameworks or whatnot they'll have reactive streams compatibility and the whole idea is that you should be able to say take like a RX Java component bolted onto an accra streams component bolted onto hopefully whatever like Netflix or Twitter open sources in the future and just have it work right you have everything communicates and you have back pressure all the way through because that's what demand right it's back pressure if you don't have demand from downstream then you can't request more from upstream than can fit in your buffer right so if there's no demand here then they'll be no demand here no demand here you sort of get this contagious back pressure right and problem is though like look at this look at these trades right everything returns unit is parameterised by type so that's cool but it's basically not really type safe but no it's basically not really functional because everything is based on side effects fortunately it's designed as a model for interoperability not a language not a trait that you implement every time you create a stream processing pipeline so that's where akka streams comes in the whole goal of ABBA streams is to build blueprints describing your stream processing pipeline that are type-safe that can be composed and that have all like their richness and power of the scala collections api so you can map over them you can filter them there are things sort of analogous to flat map and not quite I'll get into that later if there any questions but it's sort of similar to how spark is succeeded by providing something similar to the scholar collections API which we all know and love so let's go over the sort of primitives that you have available to you in accra strings right so you have a source to type parameters out and materialize right Matt's a bit complicated but out that's just what it emits that's what it publishes so it will emit 0 to n elements there won't always be the same if you create the same source strikes if it's like stock data at different points in time you'll get a different stream it's potentially unbounded because it's a stock data again as long as the New York Stock Exchange doesn't go under right let's still have stock data coming then materialize that's a bit more complicated so that is a something you get once whenever you turn this blueprint into a running stream pipeline so let's say over read file lines right take the file name as a string creates like source of string future and closable it's just an example pretty sure it doesn't exist anywhere but what that would do right it would take your file it would open like a connection or open it start reading that's where your source of fire strings comes from and then the future of closeable is say something you get when the miles open not instantly when this is created it's not not blocking then you could pose it or whatever right so that's your source that's why you get elements as your publishers and you have flows those are sort of pipeline component they have both the inlet in an outlet they both publish and subscribe so there's something you'd bolt on to a source like if you want to transform it so for example you have the source of strings right is publishing bad words you want to filter out the ones you can't show on like daytime TV you'd have a flow that just goes from string to string and then it's materialized type is unit what that means is that you don't really get anything when you materialize that because it's just like a pipeline it doesn't like do anything off to the side that would require like a future or some data that you need to access finally you have your sources you have your flows you have your sinks these consume elements in this case they consume elements of type in and they yield the value of type Matt materialized when they're well but you created so if you have a head sink right it's like equivalent to getting the head of a list but since this is from streams and not on data collections in memory what you have here is a sink of type T whatever it's consuming and that can't be a vow its type parameter I sorry but and the future of T is what you get when you create it right and that future would contain the first element push to that sick so those are for linear stream pipelines like you have a to b to c data Trevor travels in a single line it doesn't branch out it doesn't get merged so there also is a graph dsl for creating more complex systems like for example if you want to merge two streams you want to fan out two streams you want to create like custom logic it's fully supported right so you'll notice I haven't really mentioned akka before that's I mean it's a great too it's a great tool it's really great if you can like write a finite state machine that fully specifies what you want to do but then if you try and build a system only on akka you start running into like the type any to unit problem right so the whole point of back of streams at least as I see it is that you can get all the performance all like the reliability all the supervision of akka without having to at compile time lose that type safety so you actually create like a stream processing blueprint that's fully typesafe then at runtime you turn that into a system of actors that actually do the dirty work and implement the reactor stream standard so that's why I've sort of not really mentioned akka you don't really need to know that much about equity use axe streams think of your actor system as your execution context right it does all the work you don't need to know how its implemented although of course it'll be a better engineer if you do right so the example project that we're using to showcase this we're just going to scrape reddit it's we were talking about community in the ways it goes wrong before so that's a good example so write some simple types and the trait that we'll be using for this so first popular links you just take a subreddit you get a future of link listing link listing is just a sequence of links link is just an ID and a subreddit and popular comments you take a link you get a future comment listing subreddit and sequence of comments comments is just a subreddit and a body pretty simple right we're just munching the JSON behind the scenes to turn the result of the API that's whatever then you have popular subreddits it just gets you a future sequence of string of whatever subreddits are popular at that time let's go into some repple examples right so all right so let's take four first just the simplest source you can create right it just takes a vector and publishes every element in that so weird it take funny and sad we're going to run for each on that what that's doing is taking your source appending a sink to it that runs a function on every element and then run and then materializing it as just a convenience function so what we're doing we're running print line on every element of that source every element of that vector and we see funny and sad yes so that's pretty simple it doesn't really matter let's do something a bit more complicated so we have our subreddits method from before right it gets a list of popular subreddits from the reddit API it's actually a network so what we're doing here we're using the popular subreddits function returns a future of sequence of string we're wrapping that in source it's one of the many helper functions on the source object what that does is just create a source that emits a single element when that future actually completes so then we're using map concat that takes apply as a function to each element that function returns a sequence it's a way of sort of flattening out a stream that has a sequence or something that can be turned into a sequence into a stream of single elements so we're using just identity with X we already have a sequence and then again we're using run for a run for each and print line so actual popular subreddits these are like sub-forums on the reddit reddit you see funny pics advice animals they learn and of course gifts so yeah it's a pretty simple it's sort of a bridge between futures which I hope everyone's familiar with and akka streams it's very easy to interoperate with you something more complicated right so what we're going to be doing we're going to be getting a word counts for every subreddit on reddit right really simple problem it's used everywhere for stream processing really this is just to show how to use APIs dreams with something people should be familiar with so word cow sync this is doing similar to the for each sink that we saw earlier except instead of just running a function on every element it's actually folding over them and making available a few sir when it completes with the result of that fold so the future its map of string and word count or word count is just a simple type that's a map of string string and long and that's what's available when this completes and it consumes comments folds over them produces this map of string to work out right so what we're going to do we're going to create a vector of comments from two in the new subreddit hello world and cruel world one in funny hello world whatever then we're going to pipe that into word count sink and run it with run with and then on the completion of the future that we get from that we're going to just print mine let's see how that works right so as expected right we have a in the funny subreddit we have word one hello one in the new subreddit we have world too cruel one low 1 and yes so still pretty simple but sort of building up something so then what we have to do after that we have our subreddits right we want to find the popular links for every subreddit so we're going to build another pipeline for this one we have a bit of a black box it's not yet defined but we'll do that later it's called throttle this is just a pipeline stage no special machinery that just only allows n elements or one element per time unit to pass so we're passing it reddit API rate that's 500 milliseconds into the throttle we're starting with a flow of just string piping that through the file and then using map async unordered which is kind of cool to grab popular links for every subreddit and all right map a sec right so it takes a function from TD or from TD future of some other type and then just runs on every element of the stream folds the result of the future back into the stream reason it's unordered this is actually kind of important is because let's say one of the calls takes 100 milliseconds every other call takes like 10 milliseconds we don't want a backlog of calls to wait for the one out of like a thousand calls that takes say ten or a hundred times as long as the others so here we're deliberately sacrificing the property of ordering on our stream first feed and there's also a map a sink if you don't want to do that but so once again we're piping everything through every string which is a subreddit name through our throttle we're using map async unordered to hit a reddit to grab popular links and then map concat to flatten out every listening into the links on APIs and sorry I'm kind of going around a bit here but the reason we're using throttle is of course the reddit API is metered we don't want to send out all our requests at once which is what we do if we use like a naive future sequence like we have one future we get sequence of say links we grab we dispatch a future or call to get all of the comments for those links all at once right I actually tried that we get banned from reddit really quickly we do great things for a Productivity but it would not help out with this so see this pipeline here fetch links what we're going to do is a create just a source from a vector of funny sad politics and news right some subreddit names and pipe them through fetch links and then run for each again with print line so you see here we grab a link then that link contains or sorry we grab a links for subreddit each of those contains multiple links with their ID in their subreddit and then that's flattened out so each of these is a print line operation run on an element in the stream each API call results in multiple elements in the stream right there's no one to one requirement where you every element that goes into a flow has to result in one element going out of it you can filter or you can create multiple output elements from one elements so pretty simple right just grabbing funny sad and politics and that successfully so then what we want to do from those links is grab comments that we'll use to produce workouts so here it's very similar it's again piping through a throttle to avoid hitting their API too much map a sink and order it again using reddit API to popular comments to grab the popular comments for each link and then map can cat again to flatten out the listing of comments into individual comments and that what i meant to post right so once again and this is by the way the comments for yeah I know sorry it's the comments for a link on reddit where they were discussing police brutality's so there's some heated anyway responses and that's where the filter profanity flow that came in earlier that I sadly only showed the interface for was used so that's basically our pipeline right we have we showed it kind of out of order deliberately to show how you can build and compose these pieces started by just with the source of subreddits then a sink that turns comments into word counts and eventually yield zafir eventually yields the actual result of folding over all the comments and then we showed some flows that we could use to go from subreddits to links to comments so then we don't know how throttles implemented though that's one thing you can't implement using just the simple linear API reason for that is it needs to look something like this right so you have a source of ticks it produces one just like tick object every 10 milliseconds then you have whatever is your input you zip them together what happens there is you can't omit from the zip node until you have both a tee and tick a unit then you map over if throw away the UNIX it's just there to provide timing then you emit out so this entire sequence that does do some merging can be represented as a flow in and of itself but inside that flow you're doing something more complicated you need to use the graph DSL let's see how to do that right so this is our throttle takes a rate and it creates a flow as such so first you create a tick source it just creates a unit every rate in finite duration a zip node an undefined source and undefined sink that you use as like the non connected parts of that flow and then you use the flow dsl here so what we're doing we're creating a partial flow graph it's a flow that's a graph we're not everything's connected were bolting in on to the left part of the ZIF node and tick source on to the right part so again zip node can't omit until it has element on both and then just piping the out part of the zip to the undefined out that's then mapped over turned into a flow then mapped over to throw away the tick because again we don't need it so let's see how this works right we are piping a source of 1 to 10 into our throttle with 500 milliseconds and run for each we're just printing out the time that each was finished let me point out that this is not a benchmarking this is just for educational purposes so I've had it pointed out before that this is not a proper micro benchmarking technique that is correct so you can see here everything is coming out at about 500 millisecond intervals 956 457 955 etc right so the throttle works now maybe akka streams is just like mind-bogglingly slow right maybe it just takes 500 milliseconds and throttles not doing anything I figure we should at least consider that possibility so let's pipe a thousand elements just directly to a for each sink right without the actual throttle in the middle and right you can see probably a few hundred elements per millisecond like it's not slow even if you're doing nothing there's not too much impedance just by the framework so right so then you just bolt the pieces together you take a source of subreddits onto that you bolt the fetch links flow onto that you bolt the fetch comments flow then onto that you bolt your run your word count sick and that takes your strings your stream of straight or source of strings turns it into a source of links turns it into a source of comments and finally consumes that and in memory creates word counts and then there's a bit of machinery that in right results here that writes it to a tab separated value file on disk so let's run that so we're going to do is shut down the actor system quit and run so we're grabbing popular subreddits that finishes comments links right so this is going to take a few about a minute because again we're meat throttling everything is so not to be banned it does anyone have any questions is there anything that's unclear about how to use axe streams how sources sings inflows work how graphs work good question so it was what's the return type of the function I'm user I'm using to create a partial flow graph right so the return type of this is flow because first this is just sort of I couldn't tell you the type off the top of my head there's some complex machinery round types in the graph DSL but then it's a graph with only one Inlet and only one outlet so it can easily be transformed to a flow so the return type of this is a flow from t2 tuple of tea and unit and then you map over it and get rid of the units and you just have a flow of T to T any other questions let's go back to this all right let's taking too long let's just run it on a single set with it right so we're running it on the funniest subreddit which can get a tab separated value of all the words people using that we'd send it over to our data scientists they do voodoo on it we'd get market share that's how startups work right yeah so that one ran pretty quickly you can see grabbed links crab comments and then wrote the word counts let's take a look at that right see if that actually works so yeah you thought the eye to etc LOL is down here somewhere yeah so you did some basic analytics you can like make a word cloud you can create some great graphics for your next news story or whatever it's sorry it does yes there are a bit of other there are a few other considerations you need if you're building a cyclical graph basically you want to make sure that you don't deadlock or sort of deadlock it because if you have if every node in the graph right let's say you have a node a and a node B and they're just cycling data through right if they both have buffers the buffers are filled they won't be able to send any demand upstream so they won't be able to keep the data flowing even though it's a cyclical graph right that does introduce some extra complexities but it's totally possible anyone else good question any reason that you would have a circular graph right so there was an example I've seen using Kafka or some other q that you can act wear what you do is a circular graph where you get a message from Kafka you do whatever processing to it when you're done with the processing you then send an act backed Kafka which then sends you another message so you sort of include aking of messages in your graph right you explicitly model that but then you need to sort of kick start that circular graph by starting with a Act message because it won't be able to produce anything without acts so you need to sort of get the pump Prime's any other questions anything about akka streams reactive strings all right time I guess oh and yeah my boss will pay if I don't show this right contact info we're always hiring at nitro hit our careers email oh here that's that's my personal email p Kinski github and then careers that go nitro if you're interested in the opportunities and we have a booth over there alright I'm going to go hyperventilate