Stream Processing for Go
blog.wallaroolabs.com
blog.wallaroolabs.com
I'm the VP of Engineering at Wallaroo Labs. Happy to answer questions. I worked with Andy on this preview release of our Go API. Andy will be on in a while and will also be answering any questions that might come up.
We are getting ready to start work on performance tuning soon. We have a lot of work already planned there.
If you play around with the Go API, we'd love to hear from you. We need your feedback to improve our Go offering.
You can reach us via our user mailing list:
Or our IRC channel:
https://webchat.freenode.net/?channels=#wallaroo
We'd also love to do a short call with folks to discuss their Go stream processing use cases. If you email hello@wallaroolabs.com, we can set that up.
Thanks in advance!
HN was really helpful in helping us develop the new version of our Python API. We hope to get the same results with the Go API.
func (wordTotals *WordTotals) Update(word string) {
total, found := wordTotals.WordTotals[word]
if !found {
total = 0
}
wordTotals.WordTotals[word] = total + 1
}
func (wordTotals *WordTotals) Update(word string) {
wordTotals.WordTotals[word] += 1
}
Pony, however, uses the Actor model like Erlang (although the programming model is a bit different). In my opinion this is a superior model for concurrency than goroutines. I highly recommend checking it out if you have any interest in cli tools or erlang/elixir.
https://docs.google.com/spreadsheets/d/1BAiJR026ih1U8HoRw__n...
- The actor model for high-level coordination and communication
- CSP and/or FBP at the low-level, for high performance processing.
The reason for this is mainly that CSP/FBP inherently supports blocking channels between processes when processes are busy (or channels are full for FBP), thus providing implicit backpressure. This can simplify low-level programming a lot, since you don't need to worry about over-filled mailboxes or overbooked CPUs. OTOH, for communication e.g. across a cluster, this becomes problematic since you can't rely on the same level of integrity of the network connections ... meaning that the very communication-intensive nature of CSP/FBP might be less of a good fit.
Thus in my view, the actor model would fit better for communicating between services which are internally implemented e.g. with CSP/FBP.
I wrote a post about this, some time ago:
http://bionics.it/posts/flowbased-vs-erlang-message-passing
It will be interesting to see whether Pony manages to fill more than just the coordination role. Given its much higher raw-crunching performance, perhaps that is the case. I'm a little worried about the lack of implicit back-pressure (due to the actor model), but let's see.
Perhaps because of its high-performance, back-pressure won't be too costly to implement anyway? One can hope. Would be nice with a language that can encompass both programming paradigms, and be suited for the full spectrum between low-level crunching and high-level coordination and communication.
I’ve heard speculation that it really only helped scheduler efficiency and this didn’t apply all that well on SMP schedulers and has finally been removed since non-SMP builds are no longer supported.
As for Pony, their actor model doesn’t support selectively receiving messages so their queue handling may have some different trades to make there, as well as interaction with primitives like promises (which are used where Erlang would prefer a ref tagged receive clause). Spooneybarger has done work on backpressure in Pony so maybe he or other Wallaroo labs people can elaborate.
I never speak of either being better than the other. Rather, they are different. I'm an expert on backpressure in Pony, not so much in Erlang, so I'll leave it at that.
1. Head-of-line blocking caused by congestion on the single TCP connection used for message transmission between any two Erlang nodes. This can cause major problems for apps control vs. data plane design, such as Riak and Lasp. Work on Partisan (https://github.com/lasp-lang/partisan) appears to be a substantial improvement.
2. If the single remote distribution TCP connection between two nodes is broken, then the first Erlang process to send a message to the remote node is, hrrm, well, borrowed/co-opted by the BEAM VM to connect the new TCP session. IIRC that process is marked unrunnable by the scheduler until the connection is set up or perhaps there's an error. If that process is really important to the functioning of your app, for example, an important system manager in the control plane of the app, then you have a very difficult-to-diagnose latency Heisenbug to cope with.
-Scott
More generally, it doesn’t work when you have multiple schedulers, which nowadays is true for nearly all Erlang systems.
CLI should be a single binary or having a very minimal runtime that can run anywhere.
- How does the wallaroo roadmap compare to other ambitious frameworks such as Flink and Beam ?
I'm not familiar with the roadmaps for Flink and Beam so I can't comment on that. I can however, speak to our roadmap.
For Go in particular, I'm going to be starting on performance improvements very soon. Beyond that, we are waiting for feedback from folks to determine what the best course of action for the Go API is. When we released our Python API, we got a lot of excellent feedback that lead to us switching from a class based API to one that emphasized functions and decorators. We expect that similar evolution will happen with the Go API.
For Wallaroo in general, we're working on improvements to our documentation, our installation experience, Python API improvements, state object migrations (see below), as well as an expanded API to allow you to express more types of dataflows with Wallaroo. There's more beyond that, but you get the idea.
Wallaroo is definitely a younger product than Flink and Beam. We open sourced it in September of 2017 so we have a lot of work to do.
We'd love to hear from folks who are interested in being able to do stream processing in Go or Python without needing to become experts in running the JVM in production. Talking with folks helps us prioritize our work and deliver a product that can be valuable to some folks now and still more in the future as we add features.
- Tools to reason about processing/event time:
Wallaroo has a metrics UI that allows you to see throughput and within Wallaroo latency on a per-pipeline basis as well as per computation. It can be very helpful in spotting bottlenecks.
I'm interested in hearing what other tools you think would be useful. We have a few ideas but are always looking for more things we can build to make the product better.
- Is it possible to save state and restart processes with updated code?
Yes, so long as your data objects don't change. If your data objects change (for example, add a field, remove a field) then not currently. However, we started initial discussion on our approach to that this week and starting work on it in the near future.
The current solution if your data objects change is to get the state of objects out of the system before shutdown and then stream in those new values when you startup with the code that involves a schema change to your data objects. It's far from perfect but works for now until, the schema migration feature is added.
https://blog.wallaroolabs.com/2017/10/why-we-used-pony-to-wr...
TCP and Kafka are the currently supported Source and Sink types. We are in the process of improving the Kafka support.
Additionally, we are in the design phase for providing APIS in Go and Python to make it easier for Wallaroo users to provide their own sources and sinks.
We plan on adding support for only keeping part of state in memory. That's in the discussion phase. We want to talk to folks about their use cases and ideally work with them before adding it.
There are many trade-offs involved in such a feature and making those trade-offs based on informed decisions of our possible users needs is our preferred solution.
You can learn more about the existing event based windowing approach that is available now here: https://blog.wallaroolabs.com/2017/11/non-native-event-drive...
---
Can you elaborate on what specifically you are referring to as "watermarking strategies" in Flink? We are familiar with Flink's use of the Chandy Lamport algorithm but I'm not sure if that is what you are referring to.
For example, you can use the bundled MaxOutOfOrdernessTimestampExtractor to emit watermarks that lag behind the maximum timestamp seen so far by some constant amount (say, 50 ms). You could customize this behaviour by implementing the required interface, to emit a watermark, say, whenever we pass an other calendar day. You can also assign strictly-ascending watermarks which you could say is equivalent to maxOutOfOrderness = 0.
You can also choose to emit watermarks based on the content of the stream (see Punctuated Watermarks https://ci.apache.org/projects/flink/flink-docs-release-1.4/...)