[ZEPPELIN-1165 : WIP] Code-based job workflow

[cloverhearts]

What is this PR for?

Code based workflow (work in progress)

Re-implementation on this pr
#1176

if (z.getZeppelinJobStatus("execute note id", "execute paragraph id").getJobStatus().isFinished() == true)
{ z.run("execute note id", "execute paragraph id") }

or

val result = z.runSync("execute note id", "execute paragraph id")
if (result.isFinished == true)
{ println("job is done") }

or

z.run("execute note id", "execute paragraph id")
while (z.getZeppelinJobStatus("execute note id", "execute paragraph id").getJobStatus().isRunning)
{ // loop }
println("next job or done.");

Workflow process feature.
(To ensure the success of each paragraph, it is possible to run consecutively.)

Case 1

Please check the following flowchart.

Case 2

In general, when run a plurality of Paragraph, it performs Note entire run.
This is a good way to run a lot of Paragraph contained in the Note.
However, the problem occurs if the Interpreter of Paragraph different.

For Paragraph each using a different type of one of the Interpreter Note but run in sequence, the end is all different.

For example, Markdown is a very fast Interpreter.
The process is completed very quickly.
This is a problem in the sequential execution Paragraph.

This feature ensures a certain execution order Notebook with each Interpreter.

Case 3

For concurrent job in the workflow ...

If the current functional design is supposed to run at the same time, as follows
It is to share the results of the job.
But if the situation need to run the job at the same time, subject to their execution flow.

** The results will have to succeed, the following paragraph will be executed. **

What type of PR is it?

Improvement

jira

https://issues.apache.org/jira/browse/ZEPPELIN-1165

![cap 2016-07-14 15-11-07-036]

Questions:

• Does the licenses files need update? no
• Is there breaking changes for older versions? no
• Does this needs documentation? yes

[cloverhearts]

create new issue on jira
https://issues.apache.org/jira/browse/ZEPPELIN-1886

[zjffdu]

@cloverhearts This is very interesting. I have a few questions

1. Does the dynamic forms here mean more control flow (like if condition and for loop)
2. In case 2, If the markdown interpreter paragraph does not depends on the spark interpreter paragraph, we can execute them parallelly rather than sequentially.
3. I think the most important thing of workflow is to define the DAG (dependency between paragraphs). Your idea is to run the paragraphs programmatically. Would it be more intuitive to just define the DAG (Directed acyclic graph), and let the framework to run the dag automatically.
   e.g.

val flow = new JobFlow(noteId)
val note = z.getNote(noteId)
val p1 = z.getParagraph(pid1)
val p2 = z.getParagraph(pid2)
val p3 = z.getParagraph(pid3)
p3.addDependency(p2)
p2.addDependency(p1)
// run p1, p2, p3 sequentially
flow.add(p1).add(p2).add(p3).run()

4. Currently we use noteId and paragraphId, but I think these are not readable. We'd better use note name and paragraph name.

[cloverhearts]

@zjffdu
Thank you for your best advice!! :)
and Sorry, I missing the description.
actually, i made that on codebase. (dynamic form is removed.)

In fact, this feature has a dependency on Spark.
However, it is designed to be easily re-implemented in other interpreters.
There are also advantages.

Since calls can be made at any time in the code, we can use them together during analysis or in combination with external libraries.
(DAG is the same)

By default, this is not a complete implementation of the workflow.
However, I think this feature provides the basic user environment for the user to use freely.

if (z.getZeppelinJobStatus("execute note id", "execute paragraph id").getJobStatus().isFinished() == true)
{ z.run("execute note id", "execute paragraph id") }

or

val result = z.runSync("execute note id", "execute paragraph id")
if (result.isFinished == true)
{ println("job is done") }

or

z.run("execute note id", "execute paragraph id")
while (z.getZeppelinJobStatus("execute note id", "execute paragraph id").getJobStatus().isRunning)
{ // loop }
println("next job or done.");

[cloverhearts]

@zjffdu
Actually, my english not good.
If you do not mind, please give me your opinion at any time.
Thank you :)

[zjffdu]

@cloverhearts What I mean is that the code like following would be called many times by users

if (z.getZeppelinJobStatus("execute note id", "execute paragraph id").getJobStatus().isFinished() == true)
{ z.run("execute note id", "execute paragraph id") }

It is just like some code templates, so what I suggest is that we can create a high level workflow framework which use these apis internally. And for users, they just need to specify the dependency between paragraphs using this framework, they don't need to check job status like the code above.

[zjffdu]

BTW, in the first phase we can provide the high-level framework to allow user to call it programmatically, And in the second phase, it would be better to allow user to do it though drag & drop in UI.

[cloverhearts]

@zjffdu
I agree with you.
But I am a bit cautious about this part.
In fact, we've re-implemented this functionality in a variety of ways, and we've actually implemented it in the parent framework format. (Formerly PR)
If, according to your opinion, I will re-implement it, it will be a form that combines my previous PR with the current PR.
I need many people opinion.

perhaps, Woluld you give me for many opinion this about?
commiter and zeppelin users?

[zjffdu]

Thanks @cloverhearts , after reading #1176. This PR is the first phase of this feature (implement low level api for workflow), is that correct ?

[cloverhearts]

@zjffdu
Yes you are right.
And your suggestion sounds pretty attractive.

[rasehorn]

From my point of view this kind of functionality shall be provided by the core framework.
I do not have created many notebooks but what I've done always is: create one paragraph after the other to seperate data preparation from processing and visualization. So for the approach I apply it would be sufficient to execute the paragraphs in the sequence they are ordered in the notebook and this should be the default behaviour.
To support control over parallel execution of paragraphs it would be sufficient from my point of view to have a flag on each paragraph telling if this paragraph could be executed in parallel, so all subsequent paragraphs (their order within the notebook, not their ID) having this flag set could also be executed in parallel.

This is a kind defining the paragraph execution workflow implicitely without the need to program explicitely.
But again: I'm not a power user. :-)

[rasehorn]

I'm also a little bit confused what this PR really is about - the pictures above point to paragraph execution order and control but the discussion also points to Notebook execution workflows.
From my point of view the control over paragraph execution within a notebook is something different than defining a workflow for notebook execution and mixing different features leads to poor design.

Often paragraphs within notebooks depend on others and therefore they need to be executed in a certain order. I feel like this kind of paragraph execution control shall be handeled by the core framework based on settings for each paragraph within the notebook.

Additionally: In some places within the discussion the implementation of that feature on interpreter level was mentioned. It is not clear to me why the notebook workflow definition feature shall be reimplemented in different interpreters in different ways. Instead the internals of a notebook are of no interest when it is executed within a workflow - all that matters is success or failure and a definition at the workflow level what shall happen in case of a failure. So from my point of view the notebook workflow feature should also be implemented in the core code independently from the different interpreters available.

[Leemoonsoo]

z.getZeppelinJobStatus("execute note id", "execute paragraph id").getJobStatus()

How about not repeating Job, Status and omit Zeppelin (while z. represents zeppelin) in method name?
i.e. something like

z.getJob("note id", "paragraph id").getStatus()

or just

z.getJobStatus("note id", "paragraph id")

[cloverhearts]

Yes, apart from workflow, this feature is essential. (Get paragraph status)
I want to separate getZeppelinJobStatus () into a separate PR, and I want to improve the workflow by gathering this from feedback here.
And many Zeppelin users seem to want to work with a DAG type workflow outside of the interpreter.
I will put your opinions on this together and present a new alternative to this PR.

And we will separate the functions related to the workflow into other PRs.

For example, getting paragraph status, deleting paragraph output.

Thank you a lot for your opinion.

[cloverhearts]

@Leemoonsoo
Yes it seems to be good, I will make a new change.

[rasehorn]

As far as I remember another discussion the paragraph IDs will change if you export/import or copy a notebook (not sure which one applies). If that is the case the workflow will be broken after import. If the user in front of the screen is not familiar with the code and logic of the notebook, it might be difficult to fix.

What about a simple "z.wait(ordernumber or paragraphId)" function which makes the paragraph wait for the paragraph referenced by the ordernumber or id to finish successfully or cancel the paragraph execution in case of an error?

This way all paragraphs without z.wait will be executed in parallel and those calling z.wait would be executed in sequence to the ones they depend on. And additionally this kind of functionality would not be mixed with the job handling on notebook level.

[cloverhearts]

@rasehorn
Thank you for your opinion.
I have a question.
In fact, there is a little confusion in the sense of wait.
It is difficult to understand because it is a mixture of parallel and sequential.
If you enter the code in one paragraph, the paragraph will be executed sequentially from the top.
If so, you have to wait under the z.wait function, even if it is declared in parallel.
Which of the following does it mean?

Case 1

Z.run ("paragraph") // parallel run (currently support)

Z.runForWait ("paragraphID-A") // waiting for job of paragraphID-A, finished or error

Or

Case 2

Z.run ("paragraphID-A") // parallel run (currently support)
Z.wait ("paragraphID-A") // waiting for job finished or error

And would you please more explain regarding cancel?
"cancel" mean is "next job cancel"?

[rasehorn]

@cloverhearts
I think a picture and some pseudocode tells more than thousand words, so I created one.

Also: I'm only talking about the use case to ensure a certain sequence of paragraph executions when runAll is called for the notebook. If you explicitely call z.run(paragraphId) within a certain notebook after runAll() was called, you propably execute those paragraphs twice.

The easiest way to ensure a certain sequence of paragraph execution after runAll() was issued is to make the paragraphs wait for the one they depend on to finish.

Lets say we have three paragraphs.
The first one is necessary to prepare the data and define temporary tables. The second and third paragraphs depend on that data, so it does not make sense to execute them before paragraph 1 finished.
Since the last two paragraphs are in status "running" and wait in parallel for the first paragraph to finish, they will be executed in parallel.

Please see the picture

From my point of view this would be the easiest way for a ZeppelinUser to ensure a certain sequence of paragraph execution including control which paragraphs are executed in parallel.

To answer your particular question:
z.wait(idx) will cause the execution of the current paragraph to be aborted (cancelled) if the paragraph execution it was checking finished with an error.

[zjffdu]

I agree with @rasehorn that workflow execution should be done in a high level framework. User just need to define the workflow (specify the dependencies between paragraphs). I also paste one image to illustrate my current idea. In the following screenshot, we have 4 paragraphs, paragraph 1 needs to run first and paragraph 2,3,4 can be run concurrently after paragraph 1. So on each paragraph's top right area, we can allow user to specify this paragraph's dependencies. Here, paragraph_1 has no dependencies, and paragraph 2,3,4 depends on paragraph 1. After the workflow is defined (dependencies are specified), we can click the button on the top right of the note to run all the paragraphs on the note. We could also provide rest api for run this whole note.

[cloverhearts]

@rasehorn@zjffdu
Thank you very much!
I understand the function wait.
I will try to organize it again based on your opinion.
Thank you for your kind comments.

[xiufengliu]

@cloverhearts Stop??

[aviralKumar13]

When can we expect this functionality to be available in zeppelin?

[zjffdu]

@aviralKumar13 I don't think this PR is in progress now. It might be better to do that in other schedule framework. e.g airflow,

What zeppelin needs to provide is a simple use api to invoke running paragraph/notes. And this PR would be helpful on that. #3887

[aviralKumar13]

@zjffdu thanks for the info , I am looking for dependency based execution in zeppelin where in we can have one paragraph waiting for another paragraph execution to finish . Airflow can help me do the schedule , but getting response from zeppelin paragraph whether it got completed or not is not possible as on today , right ? I need to poll the spark job interface for the status of job.

[zjffdu]

@aviralKumar13 You can take a look at #3887 which provide easy api for note/paragraph execution and status polling

[aviralKumar13]

okay it looks great .. is it on track for merge to master ? Roughly when this will be available ?

[zjffdu]

I think it will be merged to master soon.

[cloverhearts]

This PR has stopped functioning.
sorry to those waiting for this PR to merge.
I think the Client API is a nice feature.
The Client API has implemented enough alternatives to this pr.