<nemosupremo>
I'm exploring Wallaroo, and I'm wondering if there are any example of working with Wallaroo with an external database for state. I currently have an analytics pipeline in Spark+Cassnadra and I'm thinking of exploring Wallaroo for performance reasons. I'd like to persist data to Cassandra (for example, minutely metrics that shouldn't persist in Wallaroo forever)
<nemosupremo>
I also need to load some external state into Wallaroo for processing as well (ex. customer configuration data)
<nemosupremo>
I've only been researching for a weekend, so I may have missed some basic concepts
<SeanTAllen>
nemosupremo: we currently do not have cassandra sources or sinks. although both would be interesting. right now, you'd need to write sources/sinks in Pony. We are about to start work on an API to allow sources and sinks to be written in Python and Go.
<SeanTAllen>
at the moment, you could load data into kafka and use the kafka source to read it in then output again to kafka using the kafka sink. or you could do the same with the TCPSource and TCPSink.
<SeanTAllen>
what sort of external state are you looking to load in? how is it used?
<nemosupremo>
yeah thats the conclusion I came too, but it seemed too hard to be true - are there no database sinks currently? Is persisting data to a database not a current goal?
<nemosupremo>
@SeanTAllen: concerning external state, its mostly client info. For example, lets say we get information about Device A. Somehow the pipeline has to know that Device A should be rolled into Customer B's dataset
<SeanTAllen>
nemosupremo: we plan on adding database sinks after we do the API for Python/Go sources/sinks. TCP and Kafka have allowed folks we've worked with so far to accomplish their goals. You could do cassandra write out from a step, its not perfect but its a workaround for now. However, that probably isn't going to address your underlying need.
<nemosupremo>
If I can output to Kafka, thats not too bad - I've once had to do something similar for Elasticsearch to make sure our ES cluster wasn't overloaded
<SeanTAllen>
A wider variety of integrations is definitely something we plan on doing. We are a small team and there's lots to do. Allowing folks to write those integrations themselves in the language of their choice is our first step towards opening that up. In general, right now, we are prioritizing the needs of clients we are working with.
<SeanTAllen>
nemosupremo: i have a lot of experience from my last job with overloading ES clusters.
<SeanTAllen>
hmmm... regarding external state. I'd need to understand your design a little more. Right now, the idea with Wallaroo is that you would stream your external state in via another pipeline that can interact with any state partition whose computations would need access to the state.
<nemosupremo>
For me, its more about getting the semantics / correctness right, which is what I'm researching. For example, if I'm computing minutely metrics I need to somehow (AFAIK) expire data from Wallaroo (so I don't hold the whole history in RAM), and then make sure I'm flushing the state correctly.
<SeanTAllen>
for example, we built a POC with a large bank that was doing risk analysis on client accounts, sometimes the parameters for given clients would change, those were streamed in as updates that would then change a field in the Client state object.
<SeanTAllen>
nemosupremo: you have complete control over your state objects and can design your own expiration of data. that said... we are going to be working on allowing state object to be "retired" if they haven't received any activity after a certain amount of time. basically a TTL.
<SeanTAllen>
re: flushing, we don't yet have the concept of a "tick" so that you can take an action based on a unit of time. (like export metrics every minute). that is planned however. currently, you would check when updating your state to see if it has been at least a minute since the last time you exported this particular metric and if yes, create a new output.
<nemosupremo>
That sounds about right wrt to the large bank PoC. The current idea I'm wrestling with is our current spark pipeline reads data from cassandra, performs an update and saves. The "Read from Cassandra" concept is hard to translate.
<SeanTAllen>
our early use cases where in the financial space where the real time systems would run for a trading day then be shut down and restarted with fresh state the next day (fairly common in that space).
<SeanTAllen>
so there are a number of features we are working on adding for other use cases.
<nemosupremo>
Once the data is inside wallaroo, then I can just use Wallaroo state and not worry about it
<SeanTAllen>
Ya.
<SeanTAllen>
So cassandra -> kafka -> wallaroo -> kafka would be your best bet for now.
<nemosupremo>
but when the application starts up, or when it sees a new device, it needs to read the historical data for that device
<SeanTAllen>
that's some possible additional infrastructure so it might not be the right time for you, depending on your needs.
<SeanTAllen>
how do you read in the historical data?
<SeanTAllen>
how does it see a new device?
<nemosupremo>
the last ~5 minutes. I don't think it would be a problem if I was starting from scratch today
<nemosupremo>
or not the last 5 minutes, it reads some stored state about the device
<SeanTAllen>
right now, the folks we are working with are starting from scratch
<SeanTAllen>
sorry, so what i meant was, on startup, how do you know what historical data needs to be read?
<SeanTAllen>
when you see a new device, how does it know there is historical data? it checks cassandra?
<nemosupremo>
wouldn't it be bad if I did a query if wallaroo state update function?
<nemosupremo>
Yes, it checks Cassandra
<nemosupremo>
Since the Spark code doesn't persist any local data
<SeanTAllen>
querying cassandra from inside the state update not optimal from a performance standpoint, but you can do it.
<nemosupremo>
I would try to add some checks to prevent from always querying cassandra, like if I already have the data in Wallaroo, no need to check cassandra
<SeanTAllen>
one more question: what does "see a new device" mean? basically, "i dont have a record in memory for this device"?
<nemosupremo>
Right
<nemosupremo>
It could be "the pipeline just started so every device is new"
<nemosupremo>
or this device was just installed, so I need to pull some infromation for this device
<nemosupremo>
I think my issue is, Cassandra is the source of truth for us and other services depend on Cassandra having up to date data
<nemosupremo>
If I was starting fresh today, then Wallaroo could have all the device state that I write out to Cassandra
<SeanTAllen>
Makes sense. You could still load all the data in from the source of truth as part of a startup and avoid the "just in time" lookup cost, but depending on constraints, that may or may not be the correct thing to do.
<SeanTAllen>
You mentioned performance problems with Spark. What are you running into?
<nemosupremo>
1. Spark needs a LOT of RAM, to a point where it doesn't make sense. This inflates our server costs (which is why its a performance issue, need to use less RAM). I've written homegrown analytics frameworks before (in Go/C++), but the team here only knows Python (PySpark).
<nemosupremo>
2. We pay a huge overhead when we process 40MB of data a minute. It takes us ~30seconds currently to process this data and we need to be under a minute
<nemosupremo>
I could probably investigate (2) some more, but I don't really care about that, I really want to see if I can solve (1)
<SeanTAllen>
makes sense
<nemosupremo>
I've written a small PoC in Go, so I know it can be done for cheaper, I just want to explore an option that lets our DS team continue to iterate in Python
<SeanTAllen>
one of the goals with wallaroo has been to keep the memory overhead of the framework low, thereby allowing more room for user data
<SeanTAllen>
same with latency of individual features
<SeanTAllen>
well, your case, of letting the DS team work in Python is one of the use cases we want to support.
<SeanTAllen>
at my previous job DS worked in Python and then had to translate it over to a JVM solution for production. There were numerous problems with that.
<nemosupremo>
Question about "autoscaling" - how does that work? Are you assuming the operator has some autoscaling group in AWS (or similar) that Wallaroo can just "take advantage of"
<nemosupremo>
or is Wallaroo a shared cluster, so you just have 100 machine running Wallaroo, that you can submit jobs too and those individual jobs can allocate as many resources as they want:
<nemosupremo>
?
<SeanTAllen>
autoscaling allows you to change the number of processes that make up the application while running. so for example, i can add an additional process or two and wallaroo will reconfigure where state is stored to take advantage of the additional capacity.
<SeanTAllen>
at the moment, it is not hooked in to any external resource managers.
<SeanTAllen>
when you start up a new wallaroo process, you can give it the address of an existing member of a cluster and it will join that cluster.
<SeanTAllen>
if you are using a resource manager like Yarn or Mesos then you can have another Wallaroo process started up within the managed resources.
<SeanTAllen>
or you could start up additional resources some place like AWS and use your normal deployment means to install the wallaroo app and fire it up.
<SeanTAllen>
There's a ton of automation tools out there, from chef and puppet et al to Mesos or Kubernetes etc.
<nemosupremo>
We use mesos, so I'm thinking what the best way to integrate. Thanks for answering my questions, I hope I didn't distract you too much
<SeanTAllen>
Our initial goal was to be agnostic so that folks can fit Wallaroo into whatever their company uses.
<SeanTAllen>
We've started looking into an "opinionated" solution for folks who dont already have a solution in place.
<SeanTAllen>
nemosupremo: we love talking to folks, learning about what their concerns are and answering their questions. Do not worry about distraction.
<SeanTAllen>
question, are you also nemosupremo on GitHub?
<nemosupremo>
yep
<SeanTAllen>
i love your profile photo
<nemosupremo>
haha its great
<SeanTAllen>
well, if you have more questions, we are here to answer them!
<nemosupremo>
A question about your Python bindings. I see Python 3 is in the pipeline - what are the roadblocks in working with Python 3? Are there serialization issues when working with Python 3 or something?
<SeanTAllen>
Given how we work with Python, there shouldnt be much to do to get it working. We embed the Python interpreter in a Wallaroo runner application and very little of the API changed from 2 to 3. The largest amount of work would be testing and ongoing maintenance.
nemosupremo has quit [Quit: My MacBook Pro has gone to sleep. ZZZzzz…]
nemosupremo has joined #wallaroo
<SeanTAllen>
nemosupremo: would you be interested in talking to a colleague of mine more about your use case? we love to hear what sort of problems folks are having. we keep them to 15 to 30 minutes and offer an awesome Wallaroo Labs shirt and thanks in return.
<nemosupremo>
Yeah, I don't mind
<SeanTAllen>
awesome! thank you.
<SeanTAllen>
if you email me your address at sean@wallaroolabs.com, i'll send an intro email
<nemosupremo>
yep done
<SeanTAllen>
intro email sent. thanks.
nemosupremo has quit [Quit: My MacBook Pro has gone to sleep. ZZZzzz…]
nemosupremo has joined #wallaroo
aturley has quit [Read error: Connection reset by peer]
aturley has joined #wallaroo
<nemosupremo>
Just wrote a Wallaroo application, and trying to run it in the docker container just leaves me with "Segmentation fault"
<nemosupremo>
I'm guessing its #960
<nemosupremo>
It wasn't. I was hardcoding the kafka broken address into my source code, like so - `wallaroo.KafkaSourceConfig(..., "kafka:9092", ...)`
<nemosupremo>
it needed to be `wallaroo.KafkaSourceConfig(..., "kafka:9092", ...)`
<nemosupremo>
This might still be #960?
<nemosupremo>
import errors (due to packages not being installed) worked fine
<nemosupremo>
but this one just dumped Seg fault
<nemosupremo>
I thought crashes were impossible in Pony (lol)
<slfritchie>
nemosupremo: They're not impossible when the FFI is used.
aturley has quit [Read error: Connection reset by peer]
aturley has joined #wallaroo
<nemosupremo>
got another segfault - is there a way to enable the dumps in docker?