Could we do this behind the scenes by writing a Receiver that publishes periodic pings?
On Tue, Apr 24, 2018 at 10:09 PM Eugene Kirpichov <kirpic...@google.com> wrote: > Kenn - I'm arguing that in Spark SDF style computation can not be > expressed at all, and neither can Beam's timers. > > Spark, unlike Flink, does not have a timer facility (only state), and as > far as I can tell its programming model has no other primitive that can map > a finite RDD into an infinite DStream - the only way to create a new > infinite DStream appears to be to write a Receiver. > > I cc'd you because I'm wondering whether you've already investigated this > when considering whether timers can be implemented on the Spark runner. > > On Tue, Apr 24, 2018 at 2:53 PM Kenneth Knowles <k...@google.com> wrote: > >> I don't think I understand what the limitations of timers are that you >> are referring to. FWIW I would say implementing other primitives like SDF >> is an explicit non-goal for Beam state & timers. >> >> I got lost at some point in this thread, but is it actually necessary >> that a bounded PCollection maps to a finite/bounded structure in Spark? >> Skimming, I'm not sure if the problem is that we can't transliterate Beam >> to Spark (this might be a good sign) or that we can't express SDF style >> computation at all (seems far-fetched, but I could be convinced). Does >> doing a lightweight analysis and just promoting some things to be some kind >> of infinite representation help? >> >> Kenn >> >> On Tue, Apr 24, 2018 at 2:37 PM Eugene Kirpichov <kirpic...@google.com> >> wrote: >> >>> Would like to revive this thread one more time. >>> >>> At this point I'm pretty certain that Spark can't support this out of >>> the box and we're gonna have to make changes to Spark. >>> >>> Holden, could you advise who would be some Spark experts (yourself >>> included :) ) who could advise what kind of Spark change would both support >>> this AND be useful to the regular Spark community (non-Beam) so that it has >>> a chance of finding support? E.g. is there any plan in Spark regarding >>> adding timers similar to Flink's or Beam's timers, maybe we could help out >>> with that? >>> >>> +Kenneth Knowles <k...@google.com> because timers suffer from the same >>> problem. >>> >>> On Thu, Apr 12, 2018 at 2:28 PM Eugene Kirpichov <kirpic...@google.com> >>> wrote: >>> >>>> (resurrecting thread as I'm back from leave) >>>> >>>> I looked at this mode, and indeed as Reuven points out it seems that it >>>> affects execution details, but doesn't offer any new APIs. >>>> Holden - your suggestions of piggybacking an unbounded-per-element SDF >>>> on top of an infinite stream would work if 1) there was just 1 element and >>>> 2) the work was guaranteed to be infinite. >>>> >>>> Unfortunately, both of these assumptions are insufficient. In >>>> particular: >>>> >>>> - 1: The SDF is applied to a PCollection; the PCollection itself may be >>>> unbounded; and the unbounded work done by the SDF happens for every >>>> element. E.g. we might have a Kafka topic on which names of Kafka topics >>>> arrive, and we may end up concurrently reading a continuously growing >>>> number of topics. >>>> - 2: The work per element is not necessarily infinite, it's just *not >>>> guaranteed to be finite* - the SDF is allowed at any moment to say >>>> "Okay, this restriction is done for real" by returning stop() from the >>>> @ProcessElement method. Continuing the Kafka example, e.g., it could do >>>> that if the topic/partition being watched is deleted. Having an infinite >>>> stream as a driver of this process would require being able to send a >>>> signal to the stream to stop itself. >>>> >>>> Is it looking like there's any other way this can be done in Spark >>>> as-is, or are we going to have to make changes to Spark to support this? >>>> >>>> On Sun, Mar 25, 2018 at 9:50 PM Holden Karau <hol...@pigscanfly.ca> >>>> wrote: >>>> >>>>> I mean the new mode is very much in the Dataset not the DStream API >>>>> (although you can use the Dataset API with the old modes too). >>>>> >>>>> On Sun, Mar 25, 2018 at 9:11 PM, Reuven Lax <re...@google.com> wrote: >>>>> >>>>>> But this new mode isn't a semantic change, right? It's moving away >>>>>> from micro batches into something that looks a lot like what Flink does - >>>>>> continuous processing with asynchronous snapshot boundaries. >>>>>> >>>>>> On Sun, Mar 25, 2018 at 9:01 PM Thomas Weise <t...@apache.org> wrote: >>>>>> >>>>>>> Hopefully the new "continuous processing mode" in Spark will enable >>>>>>> SDF implementation (and real streaming)? >>>>>>> >>>>>>> Thanks, >>>>>>> Thomas >>>>>>> >>>>>>> >>>>>>> On Sat, Mar 24, 2018 at 3:22 PM, Holden Karau <hol...@pigscanfly.ca> >>>>>>> wrote: >>>>>>> >>>>>>>> >>>>>>>> On Sat, Mar 24, 2018 at 1:23 PM Eugene Kirpichov < >>>>>>>> kirpic...@google.com> wrote: >>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>> On Fri, Mar 23, 2018, 11:17 PM Holden Karau <hol...@pigscanfly.ca> >>>>>>>>> wrote: >>>>>>>>> >>>>>>>>>> On Fri, Mar 23, 2018 at 7:00 PM Eugene Kirpichov < >>>>>>>>>> kirpic...@google.com> wrote: >>>>>>>>>> >>>>>>>>>>> On Fri, Mar 23, 2018 at 6:49 PM Holden Karau < >>>>>>>>>>> hol...@pigscanfly.ca> wrote: >>>>>>>>>>> >>>>>>>>>>>> On Fri, Mar 23, 2018 at 6:20 PM Eugene Kirpichov < >>>>>>>>>>>> kirpic...@google.com> wrote: >>>>>>>>>>>> >>>>>>>>>>>>> On Fri, Mar 23, 2018 at 6:12 PM Holden Karau < >>>>>>>>>>>>> hol...@pigscanfly.ca> wrote: >>>>>>>>>>>>> >>>>>>>>>>>>>> On Fri, Mar 23, 2018 at 5:58 PM Eugene Kirpichov < >>>>>>>>>>>>>> kirpic...@google.com> wrote: >>>>>>>>>>>>>> >>>>>>>>>>>>>>> Reviving this thread. I think SDF is a pretty big risk for >>>>>>>>>>>>>>> Spark runner streaming. Holden, is it correct that Spark >>>>>>>>>>>>>>> appears to have no >>>>>>>>>>>>>>> way at all to produce an infinite DStream from a finite RDD? >>>>>>>>>>>>>>> Maybe we can >>>>>>>>>>>>>>> somehow dynamically create a new DStream for every initial >>>>>>>>>>>>>>> restriction, >>>>>>>>>>>>>>> said DStream being obtained using a Receiver that under the >>>>>>>>>>>>>>> hood actually >>>>>>>>>>>>>>> runs the SDF? (this is of course less efficient than a >>>>>>>>>>>>>>> timer-capable runner >>>>>>>>>>>>>>> would do, and I have doubts about the fault tolerance) >>>>>>>>>>>>>>> >>>>>>>>>>>>>> So on the streaming side we could simply do it with a fixed >>>>>>>>>>>>>> number of levels on DStreams. It’s not great but it would work. >>>>>>>>>>>>>> >>>>>>>>>>>>> Not sure I understand this. Let me try to clarify what SDF >>>>>>>>>>>>> demands of the runner. Imagine the following case: a file >>>>>>>>>>>>> contains a list >>>>>>>>>>>>> of "master" Kafka topics, on which there are published additional >>>>>>>>>>>>> Kafka >>>>>>>>>>>>> topics to read. >>>>>>>>>>>>> >>>>>>>>>>>>> PCollection<String> masterTopics = >>>>>>>>>>>>> TextIO.read().from(masterTopicsFile) >>>>>>>>>>>>> PCollection<String> nestedTopics = >>>>>>>>>>>>> masterTopics.apply(ParDo(ReadFromKafkaFn)) >>>>>>>>>>>>> PCollection<String> records = >>>>>>>>>>>>> nestedTopics.apply(ParDo(ReadFromKafkaFn)) >>>>>>>>>>>>> >>>>>>>>>>>>> This exemplifies both use cases of a streaming SDF that emits >>>>>>>>>>>>> infinite output for every input: >>>>>>>>>>>>> - Applying it to a finite set of inputs (in this case to the >>>>>>>>>>>>> result of reading a text file) >>>>>>>>>>>>> - Applying it to an infinite set of inputs (i.e. having an >>>>>>>>>>>>> unbounded number of streams being read concurrently, each of the >>>>>>>>>>>>> streams >>>>>>>>>>>>> themselves is unbounded too) >>>>>>>>>>>>> >>>>>>>>>>>>> Does the multi-level solution you have in mind work for this >>>>>>>>>>>>> case? I suppose the second case is harder, so we can focus on >>>>>>>>>>>>> that. >>>>>>>>>>>>> >>>>>>>>>>>> So none of those are a splittabledofn right? >>>>>>>>>>>> >>>>>>>>>>> Not sure what you mean? ReadFromKafkaFn in these examples is a >>>>>>>>>>> splittable DoFn and we're trying to figure out how to make Spark >>>>>>>>>>> run it. >>>>>>>>>>> >>>>>>>>>>> >>>>>>>>>> Ah ok, sorry I saw that and for some reason parsed them as old >>>>>>>>>> style DoFns in my head. >>>>>>>>>> >>>>>>>>>> To effectively allow us to union back into the “same” DStream >>>>>>>>>> we’d have to end up using Sparks queue streams (or their equivalent >>>>>>>>>> custom >>>>>>>>>> source because of some queue stream limitations), which invites some >>>>>>>>>> reliability challenges. This might be at the point where I should >>>>>>>>>> send a >>>>>>>>>> diagram/some sample code since it’s a bit convoluted. >>>>>>>>>> >>>>>>>>>> The more I think about the jumps required to make the “simple” >>>>>>>>>> union approach work, the more it seems just using the statemapping >>>>>>>>>> for >>>>>>>>>> steaming is probably more reasonable. Although the state tracking in >>>>>>>>>> Spark >>>>>>>>>> can be somewhat expensive so it would probably make sense to >>>>>>>>>> benchmark to >>>>>>>>>> see if it meets our needs. >>>>>>>>>> >>>>>>>>> So the problem is, I don't think this can be made to work using >>>>>>>>> mapWithState. It doesn't allow a mapping function that emits infinite >>>>>>>>> output for an input element, directly or not. >>>>>>>>> >>>>>>>> So, provided there is an infinite input (eg pick a never ending >>>>>>>> queue stream), and each call produces a finite output, we would have an >>>>>>>> infinite number of calls. >>>>>>>> >>>>>>>>> >>>>>>>>> Dataflow and Flink, for example, had timer support even before >>>>>>>>> SDFs, and a timer can set another timer and thus end up doing an >>>>>>>>> infinite >>>>>>>>> amount of work in a fault tolerant way - so SDF could be implemented >>>>>>>>> on top >>>>>>>>> of that. But AFAIK spark doesn't have a similar feature, hence my >>>>>>>>> concern. >>>>>>>>> >>>>>>>> So we can do an inifinite queue stream which would allow us to be >>>>>>>> triggered at each interval and handle our own persistence. >>>>>>>> >>>>>>>>> >>>>>>>>> >>>>>>>>>> But these still are both DStream based rather than Dataset which >>>>>>>>>> we might want to support (depends on what direction folks take with >>>>>>>>>> the >>>>>>>>>> runners). >>>>>>>>>> >>>>>>>>>> If we wanted to do this in the dataset world looking at a custom >>>>>>>>>> sink/source would also be an option, (which is effectively what a >>>>>>>>>> custom >>>>>>>>>> queue stream like thing for dstreams requires), but the datasource >>>>>>>>>> APIs are >>>>>>>>>> a bit influx so if we ended up doing things at the edge of what’s >>>>>>>>>> allowed >>>>>>>>>> there’s a good chance we’d have to rewrite it a few times. >>>>>>>>>> >>>>>>>>>> >>>>>>>>>>>> Assuming that we have a given dstream though in Spark we can >>>>>>>>>>>> get the underlying RDD implementation for each microbatch and do >>>>>>>>>>>> our work >>>>>>>>>>>> inside of that. >>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>> More generally this does raise an important question if we >>>>>>>>>>>>>> want to target datasets instead of rdds/DStreams in which case i >>>>>>>>>>>>>> would need >>>>>>>>>>>>>> to do some more poking. >>>>>>>>>>>>>> >>>>>>>>>>>>>> >>>>>>>>>>>>>>> On Wed, Mar 14, 2018 at 10:26 PM Reuven Lax < >>>>>>>>>>>>>>> re...@google.com> wrote: >>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> How would timers be implemented? By outputing and >>>>>>>>>>>>>>>> reprocessing, the same way you proposed for SDF? >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>> i mean the timers could be inside the mappers within the >>>>>>>>>>>>>> system. Could use a singleton so if a partition is re-executed >>>>>>>>>>>>>> it doesn’t >>>>>>>>>>>>>> end up as a straggler. >>>>>>>>>>>>>> >>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> On Wed, Mar 14, 2018 at 7:25 PM Holden Karau < >>>>>>>>>>>>>>>> hol...@pigscanfly.ca> wrote: >>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> So the timers would have to be in our own code. >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>> On Wed, Mar 14, 2018 at 5:18 PM Eugene Kirpichov < >>>>>>>>>>>>>>>>> kirpic...@google.com> wrote: >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> Does Spark have support for timers? (I know it has >>>>>>>>>>>>>>>>>> support for state) >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>> On Wed, Mar 14, 2018 at 4:43 PM Reuven Lax < >>>>>>>>>>>>>>>>>> re...@google.com> wrote: >>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> Could we alternatively use a state mapping function to >>>>>>>>>>>>>>>>>>> keep track of the computation so far instead of outputting >>>>>>>>>>>>>>>>>>> V each time? >>>>>>>>>>>>>>>>>>> (also the progress so far is probably of a different type R >>>>>>>>>>>>>>>>>>> rather than V). >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> On Wed, Mar 14, 2018 at 4:28 PM Holden Karau < >>>>>>>>>>>>>>>>>>> hol...@pigscanfly.ca> wrote: >>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> So we had a quick chat about what it would take to add >>>>>>>>>>>>>>>>>>>> something like SplittableDoFns to Spark. I'd done some >>>>>>>>>>>>>>>>>>>> sketchy thinking >>>>>>>>>>>>>>>>>>>> about this last year but didn't get very far. >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> My back-of-the-envelope design was as follows: >>>>>>>>>>>>>>>>>>>> For input type T >>>>>>>>>>>>>>>>>>>> Output type V >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> Implement a mapper which outputs type (T, V) >>>>>>>>>>>>>>>>>>>> and if the computation finishes T will be populated >>>>>>>>>>>>>>>>>>>> otherwise V will be >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> For determining how long to run we'd up to either K >>>>>>>>>>>>>>>>>>>> seconds or listen for a signal on a port >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> Once we're done running we take the result and filter >>>>>>>>>>>>>>>>>>>> for the ones with T and V into seperate collections re-run >>>>>>>>>>>>>>>>>>>> until finished >>>>>>>>>>>>>>>>>>>> and then union the results >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> This is maybe not a great design but it was minimally >>>>>>>>>>>>>>>>>>>> complicated and I figured terrible was a good place to >>>>>>>>>>>>>>>>>>>> start and improve >>>>>>>>>>>>>>>>>>>> from. >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> Let me know your thoughts, especially the parts where >>>>>>>>>>>>>>>>>>>> this is worse than I remember because its been awhile >>>>>>>>>>>>>>>>>>>> since I thought about >>>>>>>>>>>>>>>>>>>> this. >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>>> -- >>>>>>>>>>>>>>>>>>>> Twitter: https://twitter.com/holdenkarau >>>>>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>>>>> -- >>>>>>>>>>>>>>>>> Twitter: https://twitter.com/holdenkarau >>>>>>>>>>>>>>>>> >>>>>>>>>>>>>>>> -- >>>>>>>>>>>>>> Twitter: https://twitter.com/holdenkarau >>>>>>>>>>>>>> >>>>>>>>>>>>> -- >>>>>>>>>>>> Twitter: https://twitter.com/holdenkarau >>>>>>>>>>>> >>>>>>>>>>> -- >>>>>>>>>> Twitter: https://twitter.com/holdenkarau >>>>>>>>>> >>>>>>>>> -- >>>>>>>> Twitter: https://twitter.com/holdenkarau >>>>>>>> >>>>>>> >>>>>>> >>>>> >>>>> >>>>> -- >>>>> Twitter: https://twitter.com/holdenkarau >>>>> >>>>
