4 ms·

Thank you for this work! Interesting results. There's an aspect of the setup which troubles me: You're building a view over transaction data in order to arrive

Thank you for this work! Interesting results. There's an aspect of the setup which troubles me: You're building a view over transaction data in order to arrive at current balances. You configure differential dataflow to reject timestamps arriving 5s after the watermark, and DD uses this to understand when a timestamp has been retired and all its effects may now be viewed. Which is very cool, and maintains internal consistency as compared to e.x. Flink, which is eventually consistent. But there's a cost here: DD must be able to reject inputs which arrive outside of their temporal window (in your setup, ~1/3 of inputs). Which is understandable, except that the system feeding DD may also have hiccups, and there's no way to guarantee all data is presented in time. Again, the broader context is building a view over transaction decisions made by another system(s). If inputs may be dropped, then the view may be inconsistent with that other system's view of the world. It seems important to call out that, while Flink is not internally consistent as you define it, it's eventually consistent answer is correct with respect to what the authoring system believes happened. The DD view is internally consistent, but may give you a different answer.

jamii 5y ago
Yeah, I touch on this near the bottom (ctrl-f 'bitemporal'). I think having multiple watermarks would be a neat solution. Kinda like the way flink sends markers through to get consistent snapshots for fault tolerance.
- jgraettinger1 5y ago
  I'm unclear on how this fully helps, though? As I understand it, it unblocks a dataflow to provide answers with respect to a user-time processed as-of a system-time, which is certainly an improvement. But you'd also have unbounded growth of tracked timestamps unless you're _also_ able to extract promises from the user that no further records at (or below) a user-time will be submitted... a promise they may or may not be able to make. Do I have that right ?
  - jamii 5y ago
    > Do I have that right? Yes, I think so. If you want to be able to handle out-of-order data, I don't think there is a way to garbage collect old time periods and still produce correct results unless you stop accepting new data for those time periods. With the differential dataflow code I wrote for this post, that cutoff point for gc is also coupled to when you can first emit results, creating a sharp tradeoff between latency of downstream results and how out-of-order upstream data is allowed to be. With bitemporal timestamps or multiple watermarks those are decoupled, so you can emit provisional results and correct them later without giving up internal consistency, and also make a totally separate decision about when to garbage collect. That doesn't remove the problem entirely, but it means that you can make decisions about latency and decisions about gc separately instead of them being controlled by one variable. With flink datastreams there are a lot of different levers to pull. The code that Vasia contributed recently waits for watermarks like differential dataflow. There is also a notion of triggers that emit early results from windowed operators, but these are not coordinated across different operators so they can result in internal inconsistency. The flink table api, as far as I can tell, mostly just ignores event-time outside of windowed aggregates. So it doesn't have to confront these tradeoffs. > the broader context is building a view over transaction decisions made by another system(s). Most the time this is fine. If upstream is: * a transactional database, we get ordered inputs * another system that has a notion of watermarks or liveness (eg spanner), we can use their guarantees to decide when to gc * a bunch of phones or a sensor network (eg mobile analytics), then we pick a cutoff that balance resource usage with data quality, and it's fine that we're not fully consistent with upstream because we're never compared against it The hard case would be when your upstream system doesn't have any timing guarantees AND you need to be fully consistent with it AND you need to handle data in event-time order. I can't think of any examples like that of the top of head. I think the answer would probably have to be either buy a lot of ram or to soft-gc old state - write it out to cheap slow storage and hope that really old inputs don't come in often enough to hurt performance. --- This has been an interesting conversation and I'll probably try to write this up to clarify my thinking. I'm not on hn often though, so if you want to continue feel free to email jamie@scattered-thoughts.net

jamii 5y ago

Yeah, I touch on this near the bottom (ctrl-f 'bitemporal'). I think having multiple watermarks would be a neat solution. Kinda like the way flink sends markers through to get consistent snapshots for fault tolerance.

jgraettinger1 5y ago
I'm unclear on how this fully helps, though? As I understand it, it unblocks a dataflow to provide answers with respect to a user-time processed as-of a system-time, which is certainly an improvement. But you'd also have unbounded growth of tracked timestamps unless you're _also_ able to extract promises from the user that no further records at (or below) a user-time will be submitted... a promise they may or may not be able to make. Do I have that right ?
- jamii 5y ago
  > Do I have that right? Yes, I think so. If you want to be able to handle out-of-order data, I don't think there is a way to garbage collect old time periods and still produce correct results unless you stop accepting new data for those time periods. With the differential dataflow code I wrote for this post, that cutoff point for gc is also coupled to when you can first emit results, creating a sharp tradeoff between latency of downstream results and how out-of-order upstream data is allowed to be. With bitemporal timestamps or multiple watermarks those are decoupled, so you can emit provisional results and correct them later without giving up internal consistency, and also make a totally separate decision about when to garbage collect. That doesn't remove the problem entirely, but it means that you can make decisions about latency and decisions about gc separately instead of them being controlled by one variable. With flink datastreams there are a lot of different levers to pull. The code that Vasia contributed recently waits for watermarks like differential dataflow. There is also a notion of triggers that emit early results from windowed operators, but these are not coordinated across different operators so they can result in internal inconsistency. The flink table api, as far as I can tell, mostly just ignores event-time outside of windowed aggregates. So it doesn't have to confront these tradeoffs. > the broader context is building a view over transaction decisions made by another system(s). Most the time this is fine. If upstream is: * a transactional database, we get ordered inputs * another system that has a notion of watermarks or liveness (eg spanner), we can use their guarantees to decide when to gc * a bunch of phones or a sensor network (eg mobile analytics), then we pick a cutoff that balance resource usage with data quality, and it's fine that we're not fully consistent with upstream because we're never compared against it The hard case would be when your upstream system doesn't have any timing guarantees AND you need to be fully consistent with it AND you need to handle data in event-time order. I can't think of any examples like that of the top of head. I think the answer would probably have to be either buy a lot of ram or to soft-gc old state - write it out to cheap slow storage and hope that really old inputs don't come in often enough to hurt performance. --- This has been an interesting conversation and I'll probably try to write this up to clarify my thinking. I'm not on hn often though, so if you want to continue feel free to email jamie@scattered-thoughts.net