[jira] [Commented] (SAMZA-353) Support assigning the same SSP to multiple tasknames

["Chris Riccomini (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SAMZA-353?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14118627#comment-14118627 ] 

Chris Riccomini commented on SAMZA-353:
---------------------------------------

bq. With global (unpartitioned) state, I think the container should definitely continue consuming the stream after the initial bootstrapping. Some jobs may have global state that changes continuously, or have periodic (e.g. hourly) state updates from an external process. It would be very disruptive if you had to keep restarting the job in order to pick up those state changes, or if you had to wait for some TTL to expire before the state was refreshed.

I agree. I would actually take this one step further, and consider how we might implement atomic swaps of shared state within a container (note: oops, I see that you mention this down below). We have an existing pattern where a DB index is built offline in Hadoop, pushed to the live site, and then atomically swapped to enable reads for it. This is the case with Voldemort's read-only stores. Even though the existing proposals come close with shared state, there's still no easy way to do an atomic swap. The nearest that I can think of is to somehow have two stores, and some control message that determines which one should be read off of at any given point. This seems ugly, though.

Related to this is the problem where you're generating a data set in Hadoop, and pushing through Kafka to your Samza job's shared state store. If you generate a feed that has a value for key A, and then future generations of the feed don't contain key A, you'll still have the value for key A forever, since it's sitting in a log-compacted Kafka topic. The only way around this, that I can see, is to have the Hadoop job intersect its keys from its latest run with the keys from the prior run, and send a key A message with a null value (a delete) to the Kafka topic. This seems ugly as well.

bq. Would a "global state" store be read-only, and only be updated by messages on the stream? If so, a job could still write to it by sending a message to the changelog stream, but the write would only take effect after a round-trip through the message broker. I think that's fine.

I had been thinking that it'd be read-write in my latest proposal, but I actually think that proposal is broken. In it, I propose having each container have a partition for the shared state store, and then having all containers except the partition's owner read from it. The problem with this approach is that it totally breaks partitioning by key, which every state store needs. The reason that it breaks this is because if container 0 sends a write for key A to its partition, and container 1 does the same, then key A will exist in two partitions. Which one should be read after the other? There's no ordering, and reading between partitions could be non-deterministic.

To fix this, you have to basically arrive at the approach that you propose, I think: a read-only store, where the writes are handled by sending messages to a stream. The trade-off with this approach is that it makes writes asynchronous, which is a little weird. I haven't fully considered the problems that this might cause.

bq. If the primary interface to the global state is a key-value interface (rather than a stream that is consumed), we may still want some way for a StreamTask to be notified when something changes in the state. For example, if the StreamTask needs to maintain some internal state that is derived from the global state, it would be good if every state change was also sent to the process() method.

Yea, this seems like it'd be useful. I think it also might fix some of the risk of having an asynchronous write to the KV store (see above), since you could be notified when your write comes through.

bq. An example of a use case that would benefit from a "global stream" model: Take the use case of a daily Hadoop job generating some global state, pushing it out through a stream, and the stream task performing a map-side join against it. While the global state is in the middle of being updated, the stream task will see some values from the previous day's state, and some values from the current day. In applications where such inconsistency is unacceptable, it would be better if each day had a separate copy of the state store, and the stream task did an atomic switchover to the new day's state when the push is complete. In a "global stream" model, a stream task could implement this atomic switchover itself, but in a "global state" model, the switchover would have to be part of the framework. And since it's a somewhat application-specific use case, we probably wouldn't want to make it part of the framework.

Yes, this use case is problematic. Even if we have "global streams", I think the actual StreamTask implementation is pretty ugly. You'd have to have either a control message topic, or a timeout that defines the edge of a store push.

bq. I think that global state should use the same message broker communication mechanism as we use for all other streams. I think it would be short-sighted to introduce a new communication mechanism (e.g. RPC to AM, or querying a remote database) for the purpose of global state. There's nothing preventing jobs from making RPC/database queries in application code if they want to, but I believe that framework-managed state should be a stream (see also comment above on continuing to consume a stream after initial bootstrapping).

I agree.

bq. Multi-subscriber partitions: I think the use case you give in the design doc (hot standby) is a good and valid one. I see great potential for Samza as a framework for maintaining materialised views of data in streams, and making it queryable by external processes (cf. SAMZA-316). However, I have been worried about the lack of high availability in such an architecture (after container failure, its partitions become unavailable for a while as they are re-bootstrapped). Thus it would be valuable to have the option of implementing hot standby (even with the caveat that you need to ensure that replicas of the same partition treat the input messages deterministically).

I agree hot standby seems desirable in the long run. I do question this implementation style, though. It seems more elegant to simply implement a hot standby by having it consume from the changelog and checkpoint topics for all StreamTasks in a container. In such a case, you don't need to have multi-subscriber streams. I really struggled to come up with any use case where multi-subscriber streams would be desirable.

bq. Regarding ordered offsets: I would prefer a solution which doesn't require offsets to be totally ordered. For example, I was thinking about implementing a MySQL system consumer (SAMZA-200) in a way that didn't have totally ordered offsets. I thought about using the MySQL GTID as offset, which consists of two parts: a UUID and a transaction number. The transaction number is sequential, but every time the master changes (failover), the transaction number sequence is reset and restarted with a new UUID. This means that in order to establish a total ordering of GTIDs, you need a lookup table on the side, telling you which UUID corresponds to which period of mastership; although there is a total ordering of transactions, it is not apparent from the GTIDs themselves.

That's really good to know. I don't think we'd had any non-ordered offset examples yet, so this is really helpful. I'd prefer more general, rather than more restrictive, all things being equal.

bq. Regarding the problem of SamzaContainer consuming the same stream at different offsets: as a suggestion, we could specify that if several tasks within the same container consume the same SSP, they will always consume at the same position. That may still require changing MessageChooser, but would probably be less intrusive than consuming at multiple offsets or requiring totally ordered offsets. For example, multi-subscriber partitions may be special-cased in MessageChooser (if one task instance chooses to process a message from a multi-subscriber partition, then all tasks which are assigned that SSP within that container must also process the message).

I thought this initially, too. The problem that I see is that it's actually impossible to enforce this. There are two cases where your StreamTasks can diverge within a single container:

# Checkpoints are per-StreamTask, so a failure could occur half-way through checkpointing all StreamTasks. Some will be left with the old SSP offset, while some with the new.
# The container count could change, which would involve moving StreamTasks from one container to another. You could end up with some StreamTasks on one offset while others are on a separate one, since we are only talking about container-level lock-step offsets.

This line of thinking is what led me to the ordered offset approach. With ordering, we can have a single offset (the oldest one), and just strip process() calls for StreamTasks that were ahead.

bq. I like the fact that global state can be shared amongst all tasks in the same container. That seems like a worthwhile optimisation. This leads me to think that there are two kinds of state stores: (a) task-local stores, whose changelog stream is only read during container startup, and which thereafter are read-writable by one particular task instance; and (b) global (container-local) stores, which continually consume the changelog stream, which are readable by all tasks in the container, but which are not writable through the key-value interface (only asynchronously writable by sending a message to the changelog stream).

Agreed. This seems pretty reasonable.

bq. Regarding "shared state store": I'm not keen on using a partition per container, because containers are supposed to be purely a unit of computational resource, and not have any semantic importance for a job. Partition-per-container would violate that principle, and would make it harder to change the number of containers used by a job. (Also, restoring from all partitions would require some kind of cross-partition ordering.)

You're correct. My most recently proposed approach seems broken in this regard. I think your proposed read-only store with asynch writes is the best we can do.

bq. Allow assigning the same SSP to multiple tasks by removing the check.

I'd rather leave this feature out for now, as it seems almost totally orthogonal to global state at this point (as you say at the end of your comment).

bq. Add configuration for global state stores, which are read-only, continually consume the changelog, and are shared between all tasks in a container.

Yes, this seems to be what I'm arriving at as well.

bq. Looking at it this way, the global state and the multi-subscriber partitions seem somewhat orthogonal. We could, for example, do only global state stores on this ticket, and multi-subscriber partitions separately on another ticket.

I agree. I was actually thinking that I'd open up a separate global state ticket, and leave this as the multi-subscriber ticket. I'm mostly interested in tackling global state right now, not multi-subscriber streams.

The two sticking points to think through seem to be:

* How to implement atomic swap.
* Are asynchronous writes a problem in shared state stores.
* Should we implement callbacks for state change on the shared stores? This seems hacky.


> Support assigning the same SSP to multiple tasknames
> ----------------------------------------------------
>
>                 Key: SAMZA-353
>                 URL: https://issues.apache.org/jira/browse/SAMZA-353
>             Project: Samza
>          Issue Type: Bug
>          Components: container
>    Affects Versions: 0.8.0
>            Reporter: Jakob Homan
>         Attachments: DESIGN-SAMZA-353-0.md, DESIGN-SAMZA-353-0.pdf
>
>
> Post SAMZA-123, it is possible to add the same SSP to multiple tasknames, although currently we check for this and error out if this is done.  We should think through the implications of having the same SSP appear in multiple tasknames and support this if it makes sense.  
> This could be used as a broadcast stream that's either added by Samza itself to each taskname, or individual groupers could do this as makes sense.  Right now the container maintains a map of SSP to TaskInstance and delivers the ssp to that task instance.  With this change, we'd need to change the map to SSP to Set[TaskInstance] and deliver the message to each TI in the set.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)