Skip to content

[ZEPPELIN-699] Add new synchronous paragraph run REST API #746

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
doanduyhai wants to merge 2 commits into from
Closed

[ZEPPELIN-699] Add new synchronous paragraph run REST API #746

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
8367acf
[ZEPPELIN-699] Add new synchronous paragraph run REST API
doanduyhai Jun 2, 2016
fb0570c
[ZEPPELIN-699] Update Notebook REST API documentation
doanduyhai Jun 2, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
55 changes: 52 additions & 3 deletions docs/rest-api/rest-notebook.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -450,12 +450,12 @@ If you work with Apache Zeppelin and find a need for an additional REST API, ple
</table>

<br/>
### Run a paragraph
### Run a paragraph asynchronously
<table class="table-configuration">
<col width="200">
<tr>
<td>Description</td>
<td>This ```POST``` method runs the paragraph by given notebook and paragraph id.
<td>This ```POST``` method runs the paragraph asynchronously by given notebook and paragraph id. This API always return SUCCESS even if the execution of the paragraph fails later because the API is asynchronous
</td>
</tr>
<tr>
Expand Down Expand Up @@ -487,6 +487,56 @@ If you work with Apache Zeppelin and find a need for an additional REST API, ple
</tr>
</table>

<br/>
### Run a paragraph synchronously
<table class="table-configuration">
<col width="200">
<tr>
<td>Description</td>
<td> This ```POST``` method runs the paragraph synchronously by given notebook and paragraph id. This API can return SUCCESS or ERROR depending on the outcome of the paragraph execution
</td>
</tr>
<tr>
<td>URL</td>
<td>```http://[zeppelin-server]:[zeppelin-port]/api/notebook/job/[notebookId]/[paragraphId]```</td>
</tr>
<tr>
<td>Success code</td>
<td>200</td>
</tr>
<tr>
<td> Fail code</td>
Copy link
Member

@felixcheung felixcheung Jun 3, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

could you please fix the extra space for this and next line?

Copy link
Contributor Author

@doanduyhai doanduyhai Jun 3, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK sure

Copy link
Member

@felixcheung felixcheung Jun 7, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

thanks, I'm referring to whitespaces in/around <td> Fail code</td>, <td> 500 </td> or <td> sample JSON input (optional, only needed when if you want to update dynamic form's value) </td> - are they intentional?

<td> 500 </td>
</tr>
<tr>
<td> sample JSON input (optional, only needed when if you want to update dynamic form's value) </td>
<td><pre>
{
"name": "name of new notebook",
"params": {
"formLabel1": "value1",
"formLabel2": "value2"
}
}</pre></td>
</tr>
<tr>
<td> sample JSON response </td>
<td><pre>{"status": "OK"}</pre></td>
</tr>
<tr>
<td> sample JSON error </td>
<td><pre>
{
"status": "INTERNAL\_SERVER\_ERROR",
"body": {
"code": "ERROR",
"type": "TEXT",
"msg": "bash: -c: line 0: unexpected EOF while looking for matching ``'\nbash: -c: line 1: syntax error: unexpected end of file\nExitValue: 2"
}
}</pre></td>
</tr>
</table>

<br/>
### Stop a paragraph
<table class="table-configuration">
Expand Down Expand Up @@ -922,4 +972,3 @@ If you work with Apache Zeppelin and find a need for an additional REST API, ple
</tr>
</tr>
</table>

81 changes: 68 additions & 13 deletions zeppelin-server/src/main/java/org/apache/zeppelin/rest/NotebookRestApi.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -567,7 +567,7 @@ public Response getNoteParagraphJobStatus(@PathParam("notebookId") String notebo
}

/**
* Run paragraph job REST API
* Run asynchronously paragraph job REST API
*
* @param message - JSON with params if user wants to update dynamic form's value
* null, empty string, empty json if user doesn't want to update
Expand All @@ -580,7 +580,7 @@ public Response getNoteParagraphJobStatus(@PathParam("notebookId") String notebo
public Response runParagraph(@PathParam("notebookId") String notebookId,
@PathParam("paragraphId") String paragraphId, String message)
throws IOException, IllegalArgumentException {
LOG.info("run paragraph job {} {} {}", notebookId, paragraphId, message);
LOG.info("run paragraph job asynchronously {} {} {}", notebookId, paragraphId, message);

Note note = notebook.getNote(notebookId);
if (note == null) {
Expand All @@ -593,22 +593,60 @@ public Response runParagraph(@PathParam("notebookId") String notebookId,
}

// handle params if presented
if (!StringUtils.isEmpty(message)) {
RunParagraphWithParametersRequest request =
gson.fromJson(message, RunParagraphWithParametersRequest.class);
Map<String, Object> paramsForUpdating = request.getParams();
if (paramsForUpdating != null) {
paragraph.settings.getParams().putAll(paramsForUpdating);
AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
note.setLastReplName(paragraph.getId());
note.persist(subject);
}
}
handleParagraphParams(message, note, paragraph);

note.run(paragraph.getId());
return new JsonResponse<>(Status.OK).build();
}

/**
* Run synchronously a paragraph REST API
*
* @param noteId - noteId
* @param paragraphId - paragraphId
* @param message - JSON with params if user wants to update dynamic form's value
* null, empty string, empty json if user doesn't want to update
*
* @return JSON with status.OK
* @throws IOException, IllegalArgumentException
*/
@POST
@Path("run/{notebookId}/{paragraphId}")
@ZeppelinApi
public Response runParagraphSynchronously(@PathParam("notebookId") String noteId,
@PathParam("paragraphId") String paragraphId,
String message) throws
IOException, IllegalArgumentException {
LOG.info("run paragraph synchronously {} {} {}", noteId, paragraphId, message);

Note note = notebook.getNote(noteId);
if (note == null) {
return new JsonResponse<>(Status.NOT_FOUND, "note not found.").build();
}

Paragraph paragraph = note.getParagraph(paragraphId);
if (paragraph == null) {
return new JsonResponse<>(Status.NOT_FOUND, "paragraph not found.").build();
}

// handle params if presented
handleParagraphParams(message, note, paragraph);

if (paragraph.getListener() == null) {
note.initializeJobListenerForParagraph(paragraph);
}

paragraph.run();
Copy link
Member

@Leemoonsoo Leemoonsoo Feb 25, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

paragraph supposed to not run in this way.
It should always run through Scheduler that Interpreter provides.

Copy link
Member

@bzz bzz Jun 5, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@doanduyhai do you plan to address this issue?


final InterpreterResult result = paragraph.getResult();

if (result.code() == InterpreterResult.Code.SUCCESS) {
return new JsonResponse<>(Status.OK, result).build();
} else {
return new JsonResponse<>(Status.INTERNAL_SERVER_ERROR, result).build();
Copy link
Member

@felixcheung felixcheung Jun 7, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

instead of hardcoding to INTERNAL_SERVER_ERROR, is there a way to map result.code() ?

Copy link
Contributor Author

@doanduyhai doanduyhai Jun 10, 2016

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well, the InterpreterResult.Code enum has only 4 distinct values right now:

  1. SUCCESS,
  2. INCOMPLETE
  3. ERROR
  4. KEEP_PREVIOUS_RESULT

Success case is handle in the if block. ERROR can be mapped to Status.INTERNAL_SERVER_ERROR. INCOMPLETE and KEEP_PREVIOUS_RESULT is Zeppelin own internal state to manage streaming from interpreter so we cannot map them to a sensible HTTP error code.

In any case, since we return the complete InterpreterResult object, we also pass the error detail back to the client

Example:

image

{
    "status": "INTERNAL_SERVER_ERROR",
    "body": {
        "code": "ERROR",
        "type": "TEXT",
        "msg": "cat: x: No such file or directory\nExitValue: 1"
    }
}

}
}

/**
* Stop(delete) paragraph job REST API
*
Expand Down Expand Up @@ -800,4 +838,21 @@ public Response search(@QueryParam("q") String queryTerm) {
return new JsonResponse<>(Status.OK, notebooksFound).build();
}


private void handleParagraphParams(String message, Note note, Paragraph paragraph)
throws IOException {
// handle params if presented
if (!StringUtils.isEmpty(message)) {
RunParagraphWithParametersRequest request =
gson.fromJson(message, RunParagraphWithParametersRequest.class);
Map<String, Object> paramsForUpdating = request.getParams();
if (paramsForUpdating != null) {
paragraph.settings.getParams().putAll(paramsForUpdating);
AuthenticationInfo subject = new AuthenticationInfo(SecurityUtils.getPrincipal());
note.setLastReplName(paragraph.getId());
note.persist(subject);
}
}
}

}
26 changes: 25 additions & 1 deletion zeppelin-zengine/src/main/java/org/apache/zeppelin/notebook/Note.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@

package org.apache.zeppelin.notebook;

import static java.lang.String.format;

import java.io.IOException;
import java.io.Serializable;
import java.util.HashMap;
Expand Down Expand Up @@ -172,6 +174,28 @@ void setInterpreterFactory(InterpreterFactory factory) {
}
}

public void initializeJobListenerForParagraph(Paragraph paragraph) {
final Note paragraphNote = paragraph.getNote();
if (paragraphNote.getId().equals(this.getId())) {
throw new IllegalArgumentException(format("The paragraph %s from note %s " +
"does not belong to note %s", paragraph.getId(), paragraphNote.getId(),
this.getId()));
}

boolean foundParagraph = false;
for (Paragraph ownParagraph : paragraphs) {
if (paragraph.getId().equals(ownParagraph.getId())) {
paragraph.setListener(this.jobListenerFactory.getParagraphJobListener(this));
foundParagraph = true;
}
}

if (!foundParagraph) {
throw new IllegalArgumentException(format("Cannot find paragraph %s " +
"from note %s", paragraph.getId(), paragraphNote.getId()));
}
}

void setJobListenerFactory(JobListenerFactory jobListenerFactory) {
this.jobListenerFactory = jobListenerFactory;
}
Expand Down Expand Up @@ -484,7 +508,7 @@ public void run(String paragraphId) {
logger.debug("New paragraph: {}", pText);
p.setEffectiveText(pText);
} else {
String intpExceptionMsg = String.format("%s",
String intpExceptionMsg = format("%s",
p.getJobName()
+ "'s Interpreter "
+ requiredReplName + " not found"
Expand Down