Outer Product about streamz HOT 4 OPEN

Can I suggest that this is an extreme case of a streaming join operation. In a streaming join operation you're going to hold on to some backlog of events in each stream and when an event occurs in the others you emit if some condition is met. Full outer product is this case when the condition is lambda *args: True and the backlog is infinite.

When thinking about streaming joins there are now questions about how to define the length of a backlog. Is it a number of elements? Is it something having to do with the data itself, like a time value?

from streamz.

Hmm ok I like that vision of the problem.
As a side note we may want to look to itertools to find new nodes to implement.

Keeping track of the number of elements is ok. The timed one would mean that we would need to track at what time the data came down and then cull the data as it became to old, which may be possible. We could even ask the user to provide a pair of functions, one which assigns a value to the data and the other which asses if the value invalidates the data.

egs

source = Stream()
source2 = Stream()
timeout_node = source.conditional_product(source2, assigning_function=time, deciding function=lambda x: (time() - x) < 60)

In this case if the data is older than a minute then we remove it.

source = Stream()
source2 = Stream()
CountClass():
    def __call__():
        self.i += 1
        return self.i
countout_node = source.conditional_product(source2, assigning_function=CountClass, deciding function=lambda x: x + 5 < CountClass.i)

source = Stream()
source2 = Stream()
outer_product = source.conditional_product(source2, assigning_function=lambda *args: True, deciding function=lambda x: x, emit_on=source)

from streamz.

As a side note we may want to look to itertools to find new nodes to implement.

Sure, I'm biased, but toolz may also have operations (like streaming joins). However, I'm also inclined to only add things as they become necessary. There is a cost to adding and maintaining functionality.

Similarly for time-based joins I would say that we shouldn't deal with it until we have a reason to. Number-of-element buffers are probably more sensible to deal with.

from streamz.

That's fair.

from streamz.