data.bythebay.io: Shinji Kim, Concord: Simple & Flexible Stream Processing on Apache Mesos
so who knows what stream processing is I think we also learned a lot from this morning talk by Jake rats who here has actually used the stream processor before okay maybe half of the room I just want you to make sure if I should focus more on the technical detail of the implementation versus the features and benefits but in general i'm going to talk about stream processing how we see stream processing in the market and what features of Concorde can really help you to do student processing and real-time application and how we are planning to develop concorde further down the road so what is stream processing we basically see stream processing as processing data in motion in the past or traditionally most of the people have been processing data at rest based on the database either running a job or pouring the database but when there's a lot of data and you want to get data results faster you might want to process the data even before I hit the database so most of the stream processors today generally sits between message queues like Kafka and databases like Cassandra or HDFS a lot of people use a stream processing to run the faster etl jobs whether you want to enrich the data if you see a user ID and match it up with the actual user name if you want to do a fast aggregation let's say you want to gain the real time click your rate / publisher if you are not network or if you want to do a filtering and deduplication of the data so you can get unique URLs that is trending odd currently so if you look at this it's generally like a faster MapReduce jobs but as you run more stream processing you will see that you'll end up running a lot of your application and core business logic on top of these streaming data my team and I from working at a large private mobile exchange network where we were processing about 10 to 15 billion transactions a day and we were running fraud detection on impressions and clicks we were also going to a very fast budget update of the advertisers so that they don't overrun on their campaigns but also every time we see someone actually downloading an app or buying something on the website we want to use those events to trigger our machine learning models so we can bit better and run our auctions better so these eventually really become like an application and like a little like micro services that are running on top of data streams and in order for you to run these services you do need a support for managing these applications like how to run clusters what happens if the job fails how do we monitor so that all the jobs are running at the same performance that we want it to be so those are a lot of the issues that we personally had with Apache storm and what we wanted to solve so in high level concord is a distributed stream processing framework we designed the concorde for high performance real-time applications that require flexibility and control currently the core of concord is written in c++ as a basis framework we have a very tight integration with apache missiles which i will go into in a minute so in concord we as the other student processing frameworks sit between data sources and data sinks we see our world as the logical layer where our users or developers program their operators or the business logic or application logic what you want to do with the data or if you want to call an api afterwards or write to database that's all totally fine and we see another world of concord in the physical layer part where which operator to run on which machine what needs to happen if what operator goes down and how to replicate data inside and those are the part the physical layer part is the part that conquered automates for you and takes care of it for you so that you can really focus on this logical layer of creating operators and managing those logic so to go into little more detail as i said we see the world of streaming applications as operators in a way you can see conquered as like a global pops up mechanism basically each operator can publish to any data stream by declaring line or subscribe to any other operator bicep specifying the name of the stream so these stream names will be just a globally unique name inside Concord and if you want to just if you say I have another operator that I want to consume words then I can just add another operator so declaring an operator is just as simple as defining a metadata where you define the name of the operator names of the input streams that you are subscribed to in ice cream and names of the output streams that you're publishing to this operator you can write it in multiple languages right now we natively support Python Ruby go java Scala and C++ each operator has a more less like a primitive API functions process record gets called every time message arrives to the operator you call produce record every time you want to omit data and you can use process timer to run up timer callback function we also support each operator with its own local caching mechanism back-to-back by rocks TV so you have your own little key value store that you can run through the latest count or aggregation that you want it to do and we currently support three different routing mechanism for messages shuffle round robin and group by so with group i we use consistent hashing so that each operator gets the dedicated key value key that it's assigned to so just like you can write each operator in different languages different operators bitten in different languages can also talk to each other inside Concord so if you have a team that is writing their etl joplin scala and wants to push that to your data science team so that they can run their own model in Python you can do that in Concord very easily this is something that very specific that we personally ran into in our previous company where we had multiple different teams having their own execution environment that they wanted to use so let's say you've created an operator and now you deploy it what happens is that we target our your operator up put it on to the cluster our scheduler along with Mason's master will schedule this as a message task to running and one of the missiles agent because we have a very tight integration with missiles we get the benefit of running a dynamic resource scheduling which means as you want to scale you just add more and mesos agent and those are resources will be available for any of the operators to consume or if you want to deploy more operators what also happens is each operator is running in its own execution environment so it's isolated if any of the operators is malfunctioning you don't have to worry about turn crashing your whole cluster or any cascading failure because it's completely isolated with cpu memory and disk it's also supervised from the missiles so if any of the job fails or node fails we can retry the job to run again and this can be configurable in terms of how many times you want to be run the job in term in in the case of failure and last but not least we have a high availability the brain of Concorde which is the Concorde scheduler along with mesas master and zookeeper are all replicated so if any of those note goes down your services will still running so now for the operator when it gets to macys agent how it runs is its own execution environment so we use the Maysles container right now you can use the upper container here as well and each executor has its own tracing engine lock-free message queue router and the Cape key value store that you can use for local cash because we come with the key value cash and end the message queue inside and the tracing engine in order for you to scale all you really have to do is add more notes you don't have to grow multiple stacks together and because of the containerized environment you we support multi-tenancy you can remotely multiple applications at the same time and last but not least you can really truly manage a dynamic topology which means if you want to add a new operator or if you want to scale out an operator you don't have to worry about your downstream operators having to restart them and because in an environment in apache storm or spark you basically have two hard we start the whole topology your whole job whereas in conquer you don't have to do that so what I mean by this is your runtime environment is very flexible so this is a very simplified version of what an ad network would do where a publisher would do so there's a bunch of raw events coming in impressions and clicks will do some count and then there's a click-through rate that we are going to put it on Cassandra estate if I want to deploy your model that calculates trending URLs all i do is i deploy another operator that listens to impression count and the the URL that comes with the event and that's it you don't have to restart or touch anything else here and this can be also applied to any of them data sources or sinks and you don't have to worry about anything that's currently running same with scaling up or done so what's great about this is that you can iterate very quickly and change things without worrying about bringing down the whole cluster or affecting other jobs that are running that's deployed by other teams which we think it is really helpful for companies that want to move fast and change things are as undergo another part that concord supports is a distributed tracing this is baked inside the framework so we have a probabilistic tracing system built in basically we paint the packet at the source about like every thousand messages and we follow through the entire computational topology and measure it the NC / operator we recently integrated this with Twitter zipkin so you can see like in a more visualized fashion which operators are running in parallel which ones are running sequentially and where there could be a potential bottleneck for you for you to actually scale moreover each operator also comes with its own metrics so latency Drupal and also hardware the utilization we currently show it to you as like a log from asos it's just available on the metro cui this really gives you the visibility inside the framework so you can actually debug things very very easily another part that we feel really proud of is a conqueror performs really well at scale we've ran a two different types of benchmark so far against storm and spark streaming I won't go into too much details but basically what we've found is that Concorde runs at a very consistent performance doesn't matter what types of job it is one thing that we found with storm was that jbm ends up taking a lot of resources we're at least ten to twenty percent which gets a little bit unmanageable at scale one thing that we found about spark streaming is that if you have a job that requires a lot of internal state management for instance if you were to do a very tight windowing like every second for like a 10-second window you're der foot goes down drastically whereas for Concord we've seen a very consistent performance against different types of jobs that we run we basically tested against seven different types of jobs these ification counting pattern matching that also happens on time windows and count window basis so I'm happy to tell you more about this if you're interested in the benchmark later so as I was talking about performance one thing that we really wanted to convey with Concord and the reason why we decided to write concord in C++ is for predictability as you scale up basically any JVM systems you do have to configure JVM it so that it works right you could configure it to run it at the current scale with current types of jobs but in order for your framework to handle many different types of jobs it really becomes a challenge it basically that's what we found there are definitely approaches of the frameworks that are trying to solve this issue on the JVM front but for us we wanted to have a very basic very simple predictable predictable system and that's what we're designing for and with mesas help the whole cluster of concord is basically abstracted as number of cpus memory and disk hence in order for you to optimize your job to run your application you can really have the fine grain for full control of your runtime of the application and with this again what we are really shooting for is for teams to run type develop their application compile a tested running deployers see how it goes and then do that whole iteration again really quickly without any downtime so one question we get always asked is the message processing guarantee currently Concorde is best effort processing what this means is that if your operator crashes or if you're no crashes your local cache will go away so we generally recommend our users either to have a check mechanism if you really care about it or most of the applications that we were looking at was something that requires or has a very high temporal value of the data because after a second what happened last second may not really matter so we basically value being faster more than complete or perfect however we are currently working on our integration with Kafka where coklat becomes our message bus so instead of using our own lock free message queue will always s Kafka and it will also tell us if any of the jobs fail so where we should pick up the messages from and eventually we should be able to support exactly once processing where we currently have a design of a transactional data store that we want to use inside Concord so with this some of the things that we are currently planning right now the at least won't guarantee with a Patrick Africa is definitely one we're currently also looking into dcos integration tcos has a great UI we've also worked with marathon before which we really like so we're really excited to do that we currently have a prebuilt connectors for Kafka kinases and Cassandra but we do get asked questions and we are based on demand we want to build more sources and sync connectors & last one at least are having more support for higher level dsl is currently in plans so in summary for Concorde Concorde stream processing application framework that you can use multiple languages for you basically see the world of stream processor processing with the models of operators talking to each other which we believe that it will allow multiple teams to operate on same streams of data and actually iterate very fast with the dynamic topology you won't really have to worry about bringing down other jobs or cluster when you're updating or scaling and which means like really the big couple way of approaching the development versus DevOps work because you should really worry about the hard application logic without having to worry about how they actually run at scale so we really wanted to separate that d by and last but not least conquer is a pretty performance at scale we basically run any events at milliseconds of latency or framework inside latency / a band for p 90 99 is about 10 milliseconds and we want to have this performance applications to run at a very high predictable manner for real time applications so if you ask me where i should or why or how i should use concorde currently we recommend concorde for use cases of real-time applications again with high temporal value that you want to process as much data as you can but not having to worry about the full completeness or correctness of the result so some of the use cases that we've talked to companies in the past include fraud detection in real time as well as doing tick data processes in financial markets as well as for a lot of real-time bidding applications so if this is interesting you should check out Concord that I oh and I'm happy to talk if you guys are looking for any specific stream processors for your needs since we've also looked into many other alternatives in the market so yep I'm happy to take questions what's your model yeah yeah so the question is how what's our approach for other systems that interfaces dream processor or interfaces comport right now all of our client api's and these operators are open source on their apache license we plan to build more ourselves but we also want to have more people to try it out and build different operators on top of it so that's all so the question is the overhead that we predict by incorporating at least once Kafka is a fairly performant system so we wouldn't say it's going to be that much of a hit yeah we've seen some pretty millinery result but we want to make sure that it runs at full scale before we talk more about it yep go ahead so the question is about language binding and you want to okay the question is if we are happy with the work that we put in for the language one day it was definitely a lot of work for sure so our transport layer how we communicate between the particles is Apache drift so Eddie drift language can be supported on top of comfort however every time we change any parts of Concord we do have to update all the API so it is a lot of work we have we have had this conversation about whether we should really do more language support but we think that is a part that that as we build have more adoption and have more people using Concorde more people can contribute and build it as a community project so go ahead you say a few words on how concord uses those to scale a job over time so the question is how we use a basis to scale job over time and when you say job over time what do you mean right now we don't do anything about around auto scaling basically we encourage you to actually see deeds that can interface time to time to see if any of them are you know hitting any bottlenecks on their latency and throughput but generally how you scale any conquered operator is by just deploying the same operator or and then we'll figure out how to basically be read all the messages amongst all the operators that are deployed does that answer your question ok any other questions yes operator you could see it that way yes right so the question is how easy it is to use the API to run your existing models basically it's basically your code you will just use the the input and input of the messages how you process it and how you want to omit the messages using the Concord API in terms of what you want to do with the model where libraries that you want to use that's all up to you so we do not force any data related api's at this moment yeah which doesn't matter which libraries you're using or what you want to run inside is completely up to you yes operators that's correct yeah so the question is if we are proprietary software at this moment all of our client api's and connectors are open-source our core that's written in C++ is currently closed source we do have plans to open source as we figure out our business but go ahead I wasn't clear did you see the operators yes yeah go ahead so the question is what happens to the state when the operator fails or as you scale out the operators basically the current state will will be lost that's why we are a best-effort at most once processing but as we integrate with Kafka you'll be able to take the state away from your Kafka and process it from wherever you left off with the messages it will manage state and where we left off on them on the messages with Kafka as well so I mean our coffee cup consumer currently works as an at least once approach so if you use that to read off from Kafka any anything that goes down it will read off from again from the wherever that it left off from the partition all right well if you guys have any questions of you over here so let me know thank you for coming