Browse Source

[feature][task] Enable zeppelin schedule a whole zeppelin note (#10434)

3.1.0-release
Eric Gao 2 years ago committed by GitHub
parent
commit
4be3b877c3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
  1. 2
      docs/docs/en/guide/task/zeppelin.md
  2. 2
      docs/docs/zh/guide/task/zeppelin.md
  3. 2
      dolphinscheduler-task-plugin/dolphinscheduler-task-zeppelin/src/main/java/org/apache/dolphinscheduler/plugin/task/zeppelin/ZeppelinParameters.java
  4. 45
      dolphinscheduler-task-plugin/dolphinscheduler-task-zeppelin/src/main/java/org/apache/dolphinscheduler/plugin/task/zeppelin/ZeppelinTask.java
  5. 37
      dolphinscheduler-task-plugin/dolphinscheduler-task-zeppelin/src/test/java/org/apache/dolphinscheduler/plugin/task/zeppelin/ZeppelinTaskTest.java
  6. 54
      dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-jupyter.ts
  7. 9
      dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-zeppelin.ts

2
docs/docs/en/guide/task/zeppelin.md

@ -21,7 +21,7 @@ it will call `Zeppelin Client API` to trigger zeppelin notebook paragraph. Click
- Failed retry interval: The time interval for resubmitting the task after a failed task. It supports drop-down and hand-filling. - Failed retry interval: The time interval for resubmitting the task after a failed task. It supports drop-down and hand-filling.
- Timeout alarm: Check the timeout alarm and timeout failure. When the task exceeds the "timeout period", an alarm email will send and the task execution will fail. - Timeout alarm: Check the timeout alarm and timeout failure. When the task exceeds the "timeout period", an alarm email will send and the task execution will fail.
- Zeppelin Note ID: The unique note id for a zeppelin notebook note. - Zeppelin Note ID: The unique note id for a zeppelin notebook note.
- Zeppelin Paragraph ID: The unique paragraph id for a zeppelin notebook paragraph. - Zeppelin Paragraph ID: The unique paragraph id for a zeppelin notebook paragraph. If you want to schedule a whole note at a time, leave this field blank.
- Zeppelin Parameters: Parameters in json format used for zeppelin dynamic form. - Zeppelin Parameters: Parameters in json format used for zeppelin dynamic form.
## Task Example ## Task Example

2
docs/docs/zh/guide/task/zeppelin.md

@ -22,7 +22,7 @@
- 超时告警:勾选超时告警、超时失败,当任务超过"超时时长"后,会发送告警邮件并且任务执行失败. - 超时告警:勾选超时告警、超时失败,当任务超过"超时时长"后,会发送告警邮件并且任务执行失败.
- 前置任务:选择当前任务的前置任务,会将被选择的前置任务设置为当前任务的上游。 - 前置任务:选择当前任务的前置任务,会将被选择的前置任务设置为当前任务的上游。
- Zeppelin Note ID:Zeppelin Note对应的唯一ID。 - Zeppelin Note ID:Zeppelin Note对应的唯一ID。
- Zepplin Paragraph:Zeppelin Paragraph对应的唯一ID。 - Zeppelin Paragraph ID:Zeppelin Paragraph对应的唯一ID。如果你想一次性调度整个note,这一栏不填即可
- Zeppelin Parameters: 用于传入Zeppelin Dynamic Form的参数。 - Zeppelin Parameters: 用于传入Zeppelin Dynamic Form的参数。
## Task Example ## Task Example

2
dolphinscheduler-task-plugin/dolphinscheduler-task-zeppelin/src/main/java/org/apache/dolphinscheduler/plugin/task/zeppelin/ZeppelinParameters.java

@ -36,7 +36,7 @@ public class ZeppelinParameters extends AbstractParameters {
@Override @Override
public boolean checkParameters() { public boolean checkParameters() {
return StringUtils.isNotEmpty(this.noteId) && StringUtils.isNotEmpty(this.paragraphId); return StringUtils.isNotEmpty(this.noteId);
} }
@Override @Override

45
dolphinscheduler-task-plugin/dolphinscheduler-task-zeppelin/src/main/java/org/apache/dolphinscheduler/plugin/task/zeppelin/ZeppelinTask.java

@ -18,6 +18,7 @@
package org.apache.dolphinscheduler.plugin.task.zeppelin; package org.apache.dolphinscheduler.plugin.task.zeppelin;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
import kong.unirest.Unirest;
import org.apache.dolphinscheduler.plugin.task.api.AbstractTaskExecutor; import org.apache.dolphinscheduler.plugin.task.api.AbstractTaskExecutor;
import org.apache.dolphinscheduler.plugin.task.api.TaskConstants; import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext; import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
@ -25,11 +26,12 @@ import org.apache.dolphinscheduler.plugin.task.api.parameters.AbstractParameters
import org.apache.dolphinscheduler.spi.utils.JSONUtils; import org.apache.dolphinscheduler.spi.utils.JSONUtils;
import org.apache.dolphinscheduler.spi.utils.PropertyUtils; import org.apache.dolphinscheduler.spi.utils.PropertyUtils;
import org.apache.zeppelin.client.ClientConfig; import org.apache.zeppelin.client.ClientConfig;
import org.apache.zeppelin.client.NoteResult;
import org.apache.zeppelin.client.ParagraphResult; import org.apache.zeppelin.client.ParagraphResult;
import org.apache.zeppelin.client.Status; import org.apache.zeppelin.client.Status;
import org.apache.zeppelin.client.ZeppelinClient; import org.apache.zeppelin.client.ZeppelinClient;
import java.util.HashMap; import java.util.HashMap;
import java.util.List;
import java.util.Map; import java.util.Map;
@ -85,11 +87,34 @@ public class ZeppelinTask extends AbstractTaskExecutor {
} }
// Submit zeppelin task // Submit zeppelin task
String resultContent;
Status status = Status.FINISHED;
if (paragraphId == null) {
NoteResult noteResult = this.zClient.executeNote(noteId, zeppelinParamsMap);
List<ParagraphResult> paragraphResultList = noteResult.getParagraphResultList();
StringBuilder resultContentBuilder = new StringBuilder();
for (ParagraphResult paragraphResult : paragraphResultList) {
resultContentBuilder.append(
String.format(
"paragraph_id: %s, paragraph_result: %s\n",
paragraphResult.getParagraphId(),
paragraphResult.getResultInText()));
status = paragraphResult.getStatus();
// we treat note execution as failure if any paragraph in the note fails
// status will be further processed in method mapStatusToExitCode below
if (status != Status.FINISHED) {
break;
}
}
resultContent = resultContentBuilder.toString();
} else {
ParagraphResult paragraphResult = this.zClient.executeParagraph(noteId, paragraphId, zeppelinParamsMap); ParagraphResult paragraphResult = this.zClient.executeParagraph(noteId, paragraphId, zeppelinParamsMap);
String resultContent = paragraphResult.getResultInText(); resultContent = paragraphResult.getResultInText();
Status status = paragraphResult.getStatus(); status = paragraphResult.getStatus();
final int exitStatusCode = mapStatusToExitCode(status); }
// Use noteId-paragraph-Id as app id // Use noteId-paragraph-Id as app id
final int exitStatusCode = mapStatusToExitCode(status);
setAppIds(String.format("%s-%s", noteId, paragraphId)); setAppIds(String.format("%s-%s", noteId, paragraphId));
setExitStatusCode(exitStatusCode); setExitStatusCode(exitStatusCode);
logger.info("zeppelin task finished with results: {}", resultContent); logger.info("zeppelin task finished with results: {}", resultContent);
@ -146,6 +171,16 @@ public class ZeppelinTask extends AbstractTaskExecutor {
super.cancelApplication(status); super.cancelApplication(status);
String noteId = this.zeppelinParameters.getNoteId(); String noteId = this.zeppelinParameters.getNoteId();
String paragraphId = this.zeppelinParameters.getParagraphId(); String paragraphId = this.zeppelinParameters.getParagraphId();
if (paragraphId == null) {
logger.info("trying terminate zeppelin task, taskId: {}, noteId: {}",
this.taskExecutionContext.getTaskInstanceId(),
noteId);
Unirest.config().defaultBaseUrl(PropertyUtils.getString(TaskConstants.ZEPPELIN_REST_URL) + "/api");
Unirest.delete("/notebook/job/{noteId}").routeParam("noteId", noteId).asJson();
logger.info("zeppelin task terminated, taskId: {}, noteId: {}",
this.taskExecutionContext.getTaskInstanceId(),
noteId);
} else {
logger.info("trying terminate zeppelin task, taskId: {}, noteId: {}, paragraphId: {}", logger.info("trying terminate zeppelin task, taskId: {}, noteId: {}, paragraphId: {}",
this.taskExecutionContext.getTaskInstanceId(), this.taskExecutionContext.getTaskInstanceId(),
noteId, noteId,
@ -157,4 +192,6 @@ public class ZeppelinTask extends AbstractTaskExecutor {
paragraphId); paragraphId);
} }
}
} }

37
dolphinscheduler-task-plugin/dolphinscheduler-task-zeppelin/src/test/java/org/apache/dolphinscheduler/plugin/task/zeppelin/ZeppelinTaskTest.java

@ -33,6 +33,7 @@ import org.apache.dolphinscheduler.spi.utils.JSONUtils;
import org.apache.zeppelin.client.ParagraphResult; import org.apache.zeppelin.client.ParagraphResult;
import org.apache.zeppelin.client.NoteResult;
import org.apache.zeppelin.client.Status; import org.apache.zeppelin.client.Status;
import org.apache.zeppelin.client.ZeppelinClient; import org.apache.zeppelin.client.ZeppelinClient;
import org.junit.Assert; import org.junit.Assert;
@ -45,7 +46,6 @@ import org.powermock.core.classloader.annotations.PowerMockIgnore;
import org.powermock.core.classloader.annotations.PrepareForTest; import org.powermock.core.classloader.annotations.PrepareForTest;
import org.powermock.modules.junit4.PowerMockRunner; import org.powermock.modules.junit4.PowerMockRunner;
import java.util.HashMap;
import java.util.Map; import java.util.Map;
@ -66,6 +66,7 @@ public class ZeppelinTaskTest {
private ZeppelinClient zClient; private ZeppelinClient zClient;
private ZeppelinTask zeppelinTask; private ZeppelinTask zeppelinTask;
private ParagraphResult paragraphResult; private ParagraphResult paragraphResult;
private NoteResult noteResult;
@Before @Before
public void before() throws Exception { public void before() throws Exception {
@ -86,7 +87,7 @@ public class ZeppelinTaskTest {
} }
@Test @Test
public void testHandleWithParagraphExecutionSucess() throws Exception { public void testHandleWithParagraphExecutionSuccess() throws Exception {
when(this.paragraphResult.getStatus()).thenReturn(Status.FINISHED); when(this.paragraphResult.getStatus()).thenReturn(Status.FINISHED);
this.zeppelinTask.handle(); this.zeppelinTask.handle();
Mockito.verify(this.zClient).executeParagraph(MOCK_NOTE_ID, Mockito.verify(this.zClient).executeParagraph(MOCK_NOTE_ID,
@ -135,6 +136,30 @@ public class ZeppelinTaskTest {
Assert.assertEquals(EXIT_CODE_FAILURE, this.zeppelinTask.getExitStatusCode()); Assert.assertEquals(EXIT_CODE_FAILURE, this.zeppelinTask.getExitStatusCode());
} }
@Test
public void testHandleWithNoteExecutionSuccess() throws Exception {
String zeppelinParametersWithNoParagraphId = buildZeppelinTaskParametersWithNoParagraphId();
TaskExecutionContext taskExecutionContext= PowerMockito.mock(TaskExecutionContext.class);
when(taskExecutionContext.getTaskParams()).thenReturn(zeppelinParametersWithNoParagraphId);
this.zeppelinTask = spy(new ZeppelinTask(taskExecutionContext));
// mock zClient and note result
this.zClient = mock(ZeppelinClient.class);
this.noteResult = mock(NoteResult.class);
// use mocked zClient in zeppelinTask
doReturn(this.zClient).when(this.zeppelinTask, "getZeppelinClient");
when(this.zClient.executeNote(any(), any(Map.class))).thenReturn(this.noteResult);
when(paragraphResult.getResultInText()).thenReturn("mock-zeppelin-paragraph-execution-result");
this.zeppelinTask.init();
when(this.paragraphResult.getStatus()).thenReturn(Status.FINISHED);
this.zeppelinTask.handle();
Mockito.verify(this.zClient).executeNote(MOCK_NOTE_ID,
(Map<String, String>) mapper.readValue(MOCK_PARAMETERS, Map.class));
Mockito.verify(this.noteResult).getParagraphResultList();
Assert.assertEquals(EXIT_CODE_SUCCESS, this.zeppelinTask.getExitStatusCode());
}
private String buildZeppelinTaskParameters() { private String buildZeppelinTaskParameters() {
ZeppelinParameters zeppelinParameters = new ZeppelinParameters(); ZeppelinParameters zeppelinParameters = new ZeppelinParameters();
zeppelinParameters.setNoteId(MOCK_NOTE_ID); zeppelinParameters.setNoteId(MOCK_NOTE_ID);
@ -143,4 +168,12 @@ public class ZeppelinTaskTest {
return JSONUtils.toJsonString(zeppelinParameters); return JSONUtils.toJsonString(zeppelinParameters);
} }
private String buildZeppelinTaskParametersWithNoParagraphId() {
ZeppelinParameters zeppelinParameters = new ZeppelinParameters();
zeppelinParameters.setNoteId(MOCK_NOTE_ID);
zeppelinParameters.setParameters(MOCK_PARAMETERS);
return JSONUtils.toJsonString(zeppelinParameters);
}
} }

54
dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-jupyter.ts

@ -80,15 +80,6 @@ export function useJupyter(model: { [field: string]: any }): IJsonItem[] {
props: { props: {
placeholder: t('project.node.jupyter_parameters_tips') placeholder: t('project.node.jupyter_parameters_tips')
} }
// validate: {
// trigger: ['input', 'blur'],
// required: false,
// validator(validate: any, value: string) {
// if (!value) {
// return new Error(t('project.node.jupyter_parameters_tips'))
// }
// }
// }
}, },
{ {
type: 'input', type: 'input',
@ -97,15 +88,6 @@ export function useJupyter(model: { [field: string]: any }): IJsonItem[] {
props: { props: {
placeholder: t('project.node.jupyter_kernel_tips') placeholder: t('project.node.jupyter_kernel_tips')
} }
// validate: {
// trigger: ['input', 'blur'],
// required: false,
// validator(validate: any, value: string) {
// if (!value) {
// return new Error(t('project.node.jupyter_kernel_tips'))
// }
// }
// }
}, },
{ {
type: 'input', type: 'input',
@ -114,15 +96,6 @@ export function useJupyter(model: { [field: string]: any }): IJsonItem[] {
props: { props: {
placeholder: t('project.node.jupyter_engine_tips') placeholder: t('project.node.jupyter_engine_tips')
} }
// validate: {
// trigger: ['input', 'blur'],
// required: false,
// validator(validate: any, value: string) {
// if (!value) {
// return new Error(t('project.node.jupyter_engine_tips'))
// }
// }
// }
}, },
{ {
type: 'input', type: 'input',
@ -131,15 +104,6 @@ export function useJupyter(model: { [field: string]: any }): IJsonItem[] {
props: { props: {
placeholder: t('project.node.jupyter_execution_timeout_tips') placeholder: t('project.node.jupyter_execution_timeout_tips')
} }
// validate: {
// trigger: ['input', 'blur'],
// required: false,
// validator(validate: any, value: string) {
// if (!value) {
// return new Error(t('project.node.jupyter_execution_timeout_tips'))
// }
// }
// }
}, },
{ {
type: 'input', type: 'input',
@ -148,15 +112,6 @@ export function useJupyter(model: { [field: string]: any }): IJsonItem[] {
props: { props: {
placeholder: t('project.node.jupyter_start_timeout_tips') placeholder: t('project.node.jupyter_start_timeout_tips')
} }
// validate: {
// trigger: ['input', 'blur'],
// required: false,
// validator(validate: any, value: string) {
// if (!value) {
// return new Error(t('project.node.jupyter_start_timeout_tips'))
// }
// }
// }
}, },
{ {
type: 'input', type: 'input',
@ -165,15 +120,6 @@ export function useJupyter(model: { [field: string]: any }): IJsonItem[] {
props: { props: {
placeholder: t('project.node.jupyter_others_tips') placeholder: t('project.node.jupyter_others_tips')
} }
// validate: {
// trigger: ['input', 'blur'],
// required: false,
// validator(validate: any, value: string) {
// if (!value) {
// return new Error(t('project.node.jupyter_others_tips'))
// }
// }
// }
}, },
...useCustomParams({ model, field: 'localParams', isSimple: false }) ...useCustomParams({ model, field: 'localParams', isSimple: false })
] ]

9
dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-zeppelin.ts

@ -45,15 +45,6 @@ export function useZeppelin(model: { [field: string]: any }): IJsonItem[] {
name: t('project.node.zeppelin_paragraph_id'), name: t('project.node.zeppelin_paragraph_id'),
props: { props: {
placeholder: t('project.node.zeppelin_paragraph_id_tips') placeholder: t('project.node.zeppelin_paragraph_id_tips')
},
validate: {
trigger: ['input', 'blur'],
required: true,
validator(validate: any, value: string) {
if (!value) {
return new Error(t('project.node.zeppelin_paragraph_id_tips'))
}
}
} }
}, },
{ {

Loading…
Cancel
Save