Thanks Xinyu, that would be a nice improvement! - Chris
On Mon, May 1, 2017 at 9:58 PM, xinyu liu <xinyuliu...@gmail.com> wrote: > Looked again at Chris's beam-samza-runner implementation. Seems > LocalApplicationRunner.run() should be asynchronous too. Current > implementation is actually using a latch to wait for the StreamProcessors > to finish, which seems unnecessary. And we can provide a waitUntilFinish() > counterpart to the user. I created > https://issues.apache.org/jira/browse/SAMZA-1252 to track it. > > Thanks, > Xinyu > > On Fri, Apr 28, 2017 at 5:55 PM, xinyu liu <xinyuliu...@gmail.com> wrote: > > > Right, option #2 seems redundant for defining streams after further > > discussion here. StreamSpec itself is flexible enough to achieve both > > static and programmatic specification of the stream. Agree it's not > > convenient for now (pretty obvious after looking at your bsr > > beam.runners.samza.wrapper), and we should provide similar predefined > > convenient wrappers for user to create the StreamSpec. In your case > > something like BoundedStreamSpec.file(....) which will generate the > system > > and serialize the data as you did. > > > > We're still thinking the callback proposed in #2 can be useful for > > requirement #6: injecting other user objects in run time, such as stores > > and metrics. To simplify the user understanding further, I think we might > > hide the ApplicationRunner and expose the StreamApplication instead, > which > > will make requirement #3 not user facing. So the API becomes like: > > > > StreamApplication app = StreamApplication.local(config) > > .init (env -> { > > env.registerStore("my-store", new MyStoreFactory()); > > env.registerMetricsReporter("my-reporte", new > > MyMetricsReporterFactory()); > > }) > > .withLifeCycleListener(myListener); > > > > app.input(BoundedStreamSpec.create("/sample/input.txt")) > > .map(...) > > .window(...) > > > > app.run(); > > > > For requirement #5, I add a .withLifeCycleListener() in the API, which > can > > trigger the callbacks with life cycle events. > > > > For #4: distribution of the jars will be what we have today using the > Yarn > > localization with a remote store like artifactory or http server. We > > discussed where to put the graph serialization. The current thinking is > to > > define a general interface which can backed by a remote store, like > Kafka, > > artifactory or http server. For Kafka, it's straightforward but we will > > have the size limit or cut it by ourselves. For the other two, we need to > > investigate whether we can easily upload jars to our artifactory and > > localizing it with Yarn. Any opinions on this? > > > > Thanks, > > Xinyu > > > > On Fri, Apr 28, 2017 at 11:34 AM, Chris Pettitt < > > cpett...@linkedin.com.invalid> wrote: > > > >> Your proposal for #1 looks good. > >> > >> I'm not quite how to reconcile the proposals for #1 and #2. In #1 you > add > >> the stream spec straight onto the runner while in #2 you do it in a > >> callback. If it is either-or, #1 looks a lot better for my purposes. > >> > >> For #4 what mechanism are you using to distribute the JARs? Can you use > >> the > >> same mechanism to distribute the serialized graph? > >> > >> On Fri, Apr 28, 2017 at 12:14 AM, xinyu liu <xinyuliu...@gmail.com> > >> wrote: > >> > >> > btw, I will get to SAMZA-1246 as soon as possible. > >> > > >> > Thanks, > >> > Xinyu > >> > > >> > On Thu, Apr 27, 2017 at 9:11 PM, xinyu liu <xinyuliu...@gmail.com> > >> wrote: > >> > > >> > > Let me try to capture the updated requirements: > >> > > > >> > > 1. Set up input streams outside StreamGraph, and treat graph > building > >> as > >> > a > >> > > library (*Fluent, Beam*). > >> > > > >> > > 2. Improve ease of use for ApplicationRunner to avoid complex > >> > > configurations such as zkCoordinator, zkCoordinationService. > >> > (*Standalone*). > >> > > Provide some programmatic way to tweak them in the API. > >> > > > >> > > 3. Clean up ApplicationRunner into a single interface (*Fluent*). We > >> can > >> > > have one or more implementations but it's hidden from the users. > >> > > > >> > > 4. Separate StreamGraph from runtime environment so it can be > >> serialized > >> > (*Beam, > >> > > Yarn*) > >> > > > >> > > 5. Better life cycle management of application, parity with > >> > > StreamProcessor (*Standalone, Beam*). Stats should include exception > >> in > >> > > case of failure (tracked in SAMZA-1246). > >> > > > >> > > 6. Support injecting user-defined objects into ApplicationRunner. > >> > > > >> > > Prateek and I iterate on the ApplilcationRunner API based on these > >> > > requirements. To support #1, we can set up input streams on the > runner > >> > > level, which returns the MessageStream and allows graph building > >> > > afterwards. The code looks like below: > >> > > > >> > > ApplicationRunner runner = ApplicationRunner.local(); > >> > > runner.input(streamSpec) > >> > > .map(..) > >> > > .window(...) > >> > > runner.run(); > >> > > > >> > > StreamSpec is the building block for setting up streams here. It can > >> be > >> > > set up in different ways: > >> > > > >> > > - Direct creation of stream spec, like runner.input(new > >> StreamSpec(id, > >> > > system, stream)) > >> > > - Load from streamId from env or config, like > >> > runner.input(runner.env(). > >> > > getStreamSpec(id)) > >> > > - Canned Spec which generates the StreamSpec with id, system and > >> stream > >> > > to minimize the configuration. For example, > CollectionSpec.create(new > >> > > ArrayList[]{1,2,3,4}), which will auto generate the system and > stream > >> in > >> > > the spec. > >> > > > >> > > To support #2, we need to be able to set up StreamSpec-related > objects > >> > and > >> > > factories programmatically in env. Suppose we have the following > >> before > >> > > runner.input(...): > >> > > > >> > > runner.setup(env /* a writable interface of env*/ -> { > >> > > env.setStreamSpec(streamId, streamSpec); > >> > > env.setSystem(systemName, systemFactory); > >> > > }) > >> > > > >> > > runner.setup(->) also provides setup for stores and other runtime > >> stuff > >> > > needed for the execution. The setup should be able to serialized to > >> > config. > >> > > For #6, I haven't figured out a good way to inject user-defined > >> objects > >> > > here yet. > >> > > > >> > > With this API, we should be able to also support #4. For remote > >> > > runner.run(), the operator user classes/lamdas in the StreamGraph > >> need to > >> > > be serialized. As today, the existing option is to serialize to a > >> stream, > >> > > either the coordinator stream or the pipeline control stream, which > >> will > >> > > have the size limit per message. Do you see RPC as an option? > >> > > > >> > > For this version of API, seems we don't need the StreamApplication > >> > wrapper > >> > > as well as exposing the StreamGraph. Do you think we are on the > right > >> > path? > >> > > > >> > > Thanks, > >> > > Xinyu > >> > > > >> > > > >> > > On Thu, Apr 27, 2017 at 6:09 AM, Chris Pettitt < > >> > > cpett...@linkedin.com.invalid> wrote: > >> > > > >> > >> That should have been: > >> > >> > >> > >> For #1, Beam doesn't have a hard requirement... > >> > >> > >> > >> On Thu, Apr 27, 2017 at 9:07 AM, Chris Pettitt < > >> cpett...@linkedin.com> > >> > >> wrote: > >> > >> > >> > >> > For #1, I doesn't have a hard requirement for any change from > >> Samza. A > >> > >> > very nice to have would be to allow the input systems to be set > up > >> at > >> > >> the > >> > >> > same time as the rest of the StreamGraph. An even nicer to have > >> would > >> > >> be to > >> > >> > do away with the callback based approach and treat graph building > >> as a > >> > >> > library, a la Beam and Flink. > >> > >> > > >> > >> > For the moment I've worked around the two pass requirement (once > >> for > >> > >> > config, once for StreamGraph) by introducing an IR layer between > >> Beam > >> > >> and > >> > >> > the Samza Fluent translation. The IR layer is convenient > >> independent > >> > of > >> > >> > this problem because it makes it easier to switch between the > >> Fluent > >> > and > >> > >> > low-level APIs. > >> > >> > > >> > >> > > >> > >> > For #4, if we had parity with StreamProcessor for lifecycle we'd > >> be in > >> > >> > great shape. One additional issue with the status call that I may > >> not > >> > >> have > >> > >> > mentioned is that it provides you no way to get at the cause of > >> > failure. > >> > >> > The StreamProcessor API does allow this via the callback. > >> > >> > > >> > >> > > >> > >> > Re. #2 and #3, I'm a big fan of getting rid of the extra > >> configuration > >> > >> > indirection you currently have to jump through (this is also > >> related > >> > to > >> > >> > system consumer configuration from #1. It makes it much easier to > >> > >> discover > >> > >> > what the configurable parameters are too, if we provide some > >> > >> programmatic > >> > >> > way to tweak them in the API - which can turn into config under > the > >> > >> hood. > >> > >> > > >> > >> > On Wed, Apr 26, 2017 at 9:20 PM, xinyu liu < > xinyuliu...@gmail.com> > >> > >> wrote: > >> > >> > > >> > >> >> Let me give a shot to summarize the requirements for > >> > ApplicationRunner > >> > >> we > >> > >> >> have discussed so far: > >> > >> >> > >> > >> >> - Support environment for passing in user-defined objects > (streams > >> > >> >> potentially) into ApplicationRunner (*Beam*) > >> > >> >> > >> > >> >> - Improve ease of use for ApplicationRunner to avoid complex > >> > >> >> configurations > >> > >> >> such as zkCoordinator, zkCoordinationService. (*Standalone*) > >> > >> >> > >> > >> >> - Clean up ApplicationRunner into a single interface (*Fluent*). > >> We > >> > can > >> > >> >> have one or more implementations but it's hidden from the users. > >> > >> >> > >> > >> >> - Separate StreamGraph from environment so it can be > serializable > >> > >> (*Beam, > >> > >> >> Yarn*) > >> > >> >> > >> > >> >> - Better life cycle management of application, including > >> > >> >> start/stop/stats (*Standalone, > >> > >> >> Beam*) > >> > >> >> > >> > >> >> > >> > >> >> One way to address 2 and 3 is to provide pre-packaged runner > using > >> > >> static > >> > >> >> factory methods, and the return type will be the > ApplicationRunner > >> > >> >> interface. So we can have: > >> > >> >> > >> > >> >> ApplicationRunner runner = ApplicationRunner.zk() / > >> > >> >> ApplicationRunner.local() > >> > >> >> / ApplicationRunner.remote() / ApplicationRunner.test(). > >> > >> >> > >> > >> >> Internally we will package the right configs and run-time > >> environment > >> > >> with > >> > >> >> the runner. For example, ApplicationRunner.zk() will define all > >> the > >> > >> >> configs > >> > >> >> needed for zk coordination. > >> > >> >> > >> > >> >> To support 1 and 4, can we pass in a lambda function in the > >> runner, > >> > and > >> > >> >> then we can run the stream graph? Like the following: > >> > >> >> > >> > >> >> ApplicationRunner.zk().env(config -> > >> > environment).run(streamGraph); > >> > >> >> > >> > >> >> Then we need a way to pass the environment into the StreamGraph. > >> This > >> > >> can > >> > >> >> be done by either adding an extra parameter to each operator, or > >> > have a > >> > >> >> getEnv() function in the MessageStream, which seems to be pretty > >> > hacky. > >> > >> >> > >> > >> >> What do you think? > >> > >> >> > >> > >> >> Thanks, > >> > >> >> Xinyu > >> > >> >> > >> > >> >> > >> > >> >> > >> > >> >> > >> > >> >> > >> > >> >> On Sun, Apr 23, 2017 at 11:01 PM, Prateek Maheshwari < > >> > >> >> pmaheshw...@linkedin.com.invalid> wrote: > >> > >> >> > >> > >> >> > Thanks for putting this together Yi! > >> > >> >> > > >> > >> >> > I agree with Jake, it does seem like there are a few too many > >> > moving > >> > >> >> parts > >> > >> >> > here. That said, the problem being solved is pretty broad, so > >> let > >> > me > >> > >> >> try to > >> > >> >> > summarize my current understanding of the requirements. Please > >> > >> correct > >> > >> >> me > >> > >> >> > if I'm wrong or missing something. > >> > >> >> > > >> > >> >> > ApplicationRunner and JobRunner first, ignoring test > environment > >> > for > >> > >> the > >> > >> >> > moment. > >> > >> >> > ApplicationRunner: > >> > >> >> > 1. Create execution plan: Same in Standalone and Yarn > >> > >> >> > 2. Create intermediate streams: Same logic but different > leader > >> > >> election > >> > >> >> > (ZK-based or pre-configured in standalone, AM in Yarn). > >> > >> >> > 3. Run jobs: In JVM in standalone. Submit to the cluster in > >> Yarn. > >> > >> >> > > >> > >> >> > JobRunner: > >> > >> >> > 1. Run the StreamProcessors: Same process in Standalone & > Test. > >> > >> Remote > >> > >> >> host > >> > >> >> > in Yarn. > >> > >> >> > > >> > >> >> > To get a single ApplicationRunner implementation, like Jake > >> > >> suggested, > >> > >> >> we > >> > >> >> > need to make leader election and JobRunner implementation > >> > pluggable. > >> > >> >> > There's still the question of whether ApplicationRunner#run > API > >> > >> should > >> > >> >> be > >> > >> >> > blocking or non-blocking. It has to be non-blocking in YARN. > We > >> > want > >> > >> it > >> > >> >> to > >> > >> >> > be blocking in standalone, but seems like the main reason is > >> ease > >> > of > >> > >> use > >> > >> >> > when launched from main(). I'd prefer making it consitently > >> > >> non-blocking > >> > >> >> > instead, esp. since in embedded standalone mode (where the > >> > processor > >> > >> is > >> > >> >> > running in another container) a blocking API would not be > >> > >> user-friendly > >> > >> >> > either. If not, we can add both run and runBlocking. > >> > >> >> > > >> > >> >> > Coming to RuntimeEnvironment, which is the least clear to me > so > >> > far: > >> > >> >> > 1. I don't think RuntimeEnvironment should be responsible for > >> > >> providing > >> > >> >> > StreamSpecs for streamIds - they can be obtained with a > >> config/util > >> > >> >> class. > >> > >> >> > The StreamProcessor should only know about logical streamIds > and > >> > the > >> > >> >> > streamId <-> actual stream mapping should happen within the > >> > >> >> > SystemProducer/Consumer/Admins provided by the > >> RuntimeEnvironment. > >> > >> >> > 2. There's also other components that the user might be > >> interested > >> > in > >> > >> >> > providing implementations of in embedded Standalone mode > (i.e., > >> not > >> > >> >> just in > >> > >> >> > tests) - MetricsRegistry and JMXServer come to mind. > >> > >> >> > 3. Most importantly, it's not clear to me who creates and > >> manages > >> > the > >> > >> >> > RuntimeEnvironment. It seems like it should be the > >> > ApplicationRunner > >> > >> or > >> > >> >> the > >> > >> >> > user because of (2) above and because StreamManager also needs > >> > >> access to > >> > >> >> > SystemAdmins for creating intermediate streams which users > might > >> > >> want to > >> > >> >> > mock. But it also needs to be passed down to the > >> StreamProcessor - > >> > >> how > >> > >> >> > would this work on Yarn? > >> > >> >> > > >> > >> >> > I think we should figure out how to integrate > RuntimeEnvironment > >> > with > >> > >> >> > ApplicationRunner before we can make a call on one vs. > multiple > >> > >> >> > ApplicationRunner implementations. If we do keep > >> > >> LocalApplicationRunner > >> > >> >> and > >> > >> >> > RemoteApplication (and TestApplicationRunner) separate, agree > >> with > >> > >> Jake > >> > >> >> > that we should remove the JobRunners and roll them up into the > >> > >> >> respective > >> > >> >> > ApplicationRunners. > >> > >> >> > > >> > >> >> > - Prateek > >> > >> >> > > >> > >> >> > On Thu, Apr 20, 2017 at 10:06 AM, Jacob Maes < > >> jacob.m...@gmail.com > >> > > > >> > >> >> wrote: > >> > >> >> > > >> > >> >> > > Thanks for the SEP! > >> > >> >> > > > >> > >> >> > > +1 on introducing these new components > >> > >> >> > > -1 on the current definition of their roles (see Design > >> feedback > >> > >> >> below) > >> > >> >> > > > >> > >> >> > > *Design* > >> > >> >> > > > >> > >> >> > > - If LocalJobRunner and RemoteJobRunner handle the > >> different > >> > >> >> methods > >> > >> >> > of > >> > >> >> > > launching a Job, what additional value do the different > >> types > >> > of > >> > >> >> > > ApplicationRunner and RuntimeEnvironment provide? It > seems > >> > like > >> > >> a > >> > >> >> red > >> > >> >> > > flag > >> > >> >> > > that all 3 would need to change from environment to > >> > >> environment. It > >> > >> >> > > indicates that they don't have proper modularity. The > >> > >> >> > > call-sequence-figures > >> > >> >> > > support this; LocalApplicationRunner and > >> > RemoteApplicationRunner > >> > >> >> make > >> > >> >> > > the > >> > >> >> > > same calls and the diagram only varies after > >> jobRunner.start() > >> > >> >> > > - As far as I can tell, the only difference between Local > >> and > >> > >> >> Remote > >> > >> >> > > ApplicationRunner is that one is blocking and the other > is > >> > >> >> > > non-blocking. If > >> > >> >> > > that's all they're for then either the names should be > >> changed > >> > >> to > >> > >> >> > > reflect > >> > >> >> > > this, or they should be combined into one > ApplicationRunner > >> > and > >> > >> >> just > >> > >> >> > > expose > >> > >> >> > > separate methods for run() and runBlocking() > >> > >> >> > > - There isn't much detail on why the main() methods for > >> > >> >> Local/Remote > >> > >> >> > > have such different implementations, how they receive the > >> > >> >> Application > >> > >> >> > > (direct vs config), and concretely how the deployment > >> scripts, > >> > >> if > >> > >> >> any, > >> > >> >> > > should interact with them. > >> > >> >> > > > >> > >> >> > > > >> > >> >> > > *Style* > >> > >> >> > > > >> > >> >> > > - nit: None of the 11 uses of the word "actual" in the > doc > >> are > >> > >> >> > > *actually* > >> > >> >> > > needed. :-) > >> > >> >> > > - nit: Colors of the runtime blocks in the diagrams are > >> > >> >> unconventional > >> > >> >> > > and a little distracting. Reminds me of nai won bao. Now > >> I'm > >> > >> >> hungry. > >> > >> >> > :-) > >> > >> >> > > - Prefer the name "ExecutionEnvironment" over > >> > >> "RuntimeEnvironment". > >> > >> >> > The > >> > >> >> > > term "execution environment" is used > >> > >> >> > > - The code comparisons for the ApplicationRunners are not > >> > >> >> > apples-apples. > >> > >> >> > > The local runner example is an application that USES the > >> local > >> > >> >> runner. > >> > >> >> > > The > >> > >> >> > > remote runner example is the just the runner code itself. > >> So, > >> > >> it's > >> > >> >> not > >> > >> >> > > readily apparent that we're comparing the main() methods > >> and > >> > not > >> > >> >> the > >> > >> >> > > application itself. > >> > >> >> > > > >> > >> >> > > > >> > >> >> > > On Mon, Apr 17, 2017 at 5:02 PM, Yi Pan < > nickpa...@gmail.com> > >> > >> wrote: > >> > >> >> > > > >> > >> >> > > > Made some updates to clarify the role and functions of > >> > >> >> > RuntimeEnvironment > >> > >> >> > > > in SEP-2. > >> > >> >> > > > > >> > >> >> > > > On Fri, Apr 14, 2017 at 9:30 AM, Yi Pan < > >> nickpa...@gmail.com> > >> > >> >> wrote: > >> > >> >> > > > > >> > >> >> > > > > Hi, everyone, > >> > >> >> > > > > > >> > >> >> > > > > In light of new features such as fluent API and > standalone > >> > that > >> > >> >> > > introduce > >> > >> >> > > > > new deployment / application launch models in Samza, I > >> > created > >> > >> a > >> > >> >> new > >> > >> >> > > > SEP-2 > >> > >> >> > > > > to address the new use cases. SEP-2 link: > >> > https://cwiki.apache > >> > >> . > >> > >> >> > > > > org/confluence/display/SAMZA/ > SEP-2%3A+ApplicationRunner+ > >> > Design > >> > >> >> > > > > > >> > >> >> > > > > Please take a look and give feedbacks! > >> > >> >> > > > > > >> > >> >> > > > > Thanks! > >> > >> >> > > > > > >> > >> >> > > > > -Yi > >> > >> >> > > > > > >> > >> >> > > > > >> > >> >> > > > >> > >> >> > > >> > >> >> > >> > >> > > >> > >> > > >> > >> > >> > > > >> > > > >> > > >> > > > > >
