need help understanding map-reduce example

[ydewit]

Trying to think through how to "design" a data flow with hydroflow and nothing better than look at the examples.

I do understand the whole mechanics, but have specific questions around the following massaged flow:

let ids = workers.members();

process
/*1*/  .source_iter(q!(vec!["abc", "abc", "xyz", "abc"])) 
  .map(q!(|s| s.to_string())) // Stream<String, Process<Leader>, Unbounded, TotalOrder>
////// round_robin_bincode(..)
  .enumerate() // Stream<(usize, String), Process<Leader>, Unbounded, TotalOrder>
  .map(q!(|(i, w)| (ids[i % ids.len()], w))) // Stream<(ClusterId<Worker>, String), Process<Leader>, Unbounded, TotalOrder>
  .send_bincode(other) // Stream<String, Cluster<'_, Worker>, Unbounded>
////// end: round_robin_bincode()
/*2*/  .map(q!(|string| (string, ()))) // Stream<(String, ()), Cluster<'a, Worker>, Unbounded> 
  .enumerate()
/*3*/  .map(q!(|(i, w)| (ids[i % ids.len()], w))) // Stream<(ClusterId<Worker>, str), Cluster<Worker>, 
  .send_bincode(other)
/*4*/  .tick_batch(&cluster.tick())
  .fold_keyed(q!(|| 0), q!(|count, _| *count += 1))
  .inspect(q!(|(string, count)| println!("partition count: {} - {}", string, count)))
/*5*/  .all_ticks()
  .send_bincode(other)
  .map(q!(|(_, b)| b))
  .tick_batch(&process.tick())
/*6*/  .persist()
  .reduce_keyed_commutative(q!(|total, count| *total += count))
/*7*/  .all_ticks()
  .for_each(q!(|(string, count)| println!("{}: {}", string, count)));

1. I guess the Stream here is Unbounded because we cannot "see" that it is an iterator over a static list?
2. Why do we need the (String, ()) type here?
3. Does send_bincode to a Cluster always expects a stream of (ClusterId, Type)? I don't see that (ClusterId, _) type in the operator self.
4. Do we mainly tick_batch() to make sure we can use the Bounded stream operators? In other words, if we had an operator that worked in Unbounded stream then we would not need to call tick_batch()?
5. Does this mean that each worker will advance and process all ticks before moving forward? Does that mean workers will not send anything to the leader until all ticks are done processing? Or it just means that the event loop will advance until there are no more ticks and all the stream values have been send to the leader already?
6. Why do we need to persist here?
7. (same question as 4.)

[ydewit]

Re: 3, I see that can_send.rs resolves my question: it calculates the in and out types when crossing locations.

impl<'a, P1, C2> CanSend<'a, Cluster<'a, C2>> for Process<'a, P1> {
    type In<T> = (ClusterId<C2>, T);
    type Out<T> = T;
...