Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
apache
DolphinScheduler
提交
5c1b3f0e
DolphinScheduler
项目概览
apache
/
DolphinScheduler
上一次同步 接近 2 年
通知
707
Star
9572
Fork
3514
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
DolphinScheduler
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
5c1b3f0e
编写于
6月 21, 2019
作者:
journey2018
浏览文件
操作
浏览文件
下载
电子邮件补丁
差异文件
qianfan task result add judge
上级
421d0219
变更
5
隐藏空白更改
内联
并排
Showing
5 changed file
with
113 addition
and
2 deletion
+113
-2
escheduler-common/src/main/java/cn/escheduler/common/Constants.java
...-common/src/main/java/cn/escheduler/common/Constants.java
+5
-0
escheduler-common/src/main/java/cn/escheduler/common/enums/TaskRecordStatus.java
...ain/java/cn/escheduler/common/enums/TaskRecordStatus.java
+35
-0
escheduler-common/src/main/java/cn/escheduler/common/enums/TaskType.java
...on/src/main/java/cn/escheduler/common/enums/TaskType.java
+7
-1
escheduler-dao/src/main/java/cn/escheduler/dao/TaskRecordDao.java
...er-dao/src/main/java/cn/escheduler/dao/TaskRecordDao.java
+46
-1
escheduler-server/src/main/java/cn/escheduler/server/worker/runner/TaskScheduleThread.java
...n/escheduler/server/worker/runner/TaskScheduleThread.java
+20
-0
未找到文件。
escheduler-common/src/main/java/cn/escheduler/common/Constants.java
浏览文件 @
5c1b3f0e
...
@@ -252,6 +252,11 @@ public final class Constants {
...
@@ -252,6 +252,11 @@ public final class Constants {
*/
*/
public
static
final
String
YYYY_MM_DD_HH_MM_SS
=
"yyyy-MM-dd HH:mm:ss"
;
public
static
final
String
YYYY_MM_DD_HH_MM_SS
=
"yyyy-MM-dd HH:mm:ss"
;
/**
* date format of yyyyMMdd
*/
public
static
final
String
YYYYMMDD
=
"yyyyMMdd"
;
/**
/**
* date format of yyyyMMddHHmmss
* date format of yyyyMMddHHmmss
*/
*/
...
...
escheduler-common/src/main/java/cn/escheduler/common/enums/TaskRecordStatus.java
0 → 100644
浏览文件 @
5c1b3f0e
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package
cn.escheduler.common.enums
;
/**
* task record status
*
*/
public
enum
TaskRecordStatus
{
/**
* status:
* 0 sucess
* 1 failure
* 2 exception
*/
SUCCESS
,
FAILURE
,
EXCEPTION
}
escheduler-common/src/main/java/cn/escheduler/common/enums/TaskType.java
浏览文件 @
5c1b3f0e
...
@@ -30,5 +30,11 @@ public enum TaskType {
...
@@ -30,5 +30,11 @@ public enum TaskType {
* 6 PYTHON
* 6 PYTHON
* 7 DEPENDENT
* 7 DEPENDENT
*/
*/
SHELL
,
SQL
,
SUB_PROCESS
,
PROCEDURE
,
MR
,
SPARK
,
PYTHON
,
DEPENDENT
SHELL
,
SQL
,
SUB_PROCESS
,
PROCEDURE
,
MR
,
SPARK
,
PYTHON
,
DEPENDENT
;
public
static
boolean
typeIsNormalTask
(
String
typeName
)
{
TaskType
taskType
=
TaskType
.
valueOf
(
typeName
);
return
!(
taskType
==
TaskType
.
SUB_PROCESS
||
taskType
==
TaskType
.
DEPENDENT
);
}
}
}
escheduler-dao/src/main/java/cn/escheduler/dao/TaskRecordDao.java
浏览文件 @
5c1b3f0e
...
@@ -17,6 +17,8 @@
...
@@ -17,6 +17,8 @@
package
cn.escheduler.dao
;
package
cn.escheduler.dao
;
import
cn.escheduler.common.Constants
;
import
cn.escheduler.common.Constants
;
import
cn.escheduler.common.enums.TaskRecordStatus
;
import
cn.escheduler.common.utils.CollectionUtils
;
import
cn.escheduler.common.utils.DateUtils
;
import
cn.escheduler.common.utils.DateUtils
;
import
cn.escheduler.dao.model.TaskRecord
;
import
cn.escheduler.dao.model.TaskRecord
;
import
org.apache.commons.configuration.Configuration
;
import
org.apache.commons.configuration.Configuration
;
...
@@ -28,6 +30,7 @@ import org.slf4j.LoggerFactory;
...
@@ -28,6 +30,7 @@ import org.slf4j.LoggerFactory;
import
java.sql.*
;
import
java.sql.*
;
import
java.util.ArrayList
;
import
java.util.ArrayList
;
import
java.util.Collections
;
import
java.util.List
;
import
java.util.List
;
import
java.util.Map
;
import
java.util.Map
;
...
@@ -43,7 +46,7 @@ public class TaskRecordDao {
...
@@ -43,7 +46,7 @@ public class TaskRecordDao {
/**
/**
*
加载配置文件
*
load conf file
*/
*/
private
static
Configuration
conf
;
private
static
Configuration
conf
;
...
@@ -56,6 +59,14 @@ public class TaskRecordDao {
...
@@ -56,6 +59,14 @@ public class TaskRecordDao {
}
}
}
}
/**
* get task record flag
* @return
*/
public
static
boolean
getTaskRecordFlag
(){
return
conf
.
getBoolean
(
Constants
.
TASK_RECORD_FLAG
);
}
/**
/**
* create connection
* create connection
* @return
* @return
...
@@ -253,4 +264,38 @@ public class TaskRecordDao {
...
@@ -253,4 +264,38 @@ public class TaskRecordDao {
}
}
return
recordList
;
return
recordList
;
}
}
/**
* according to procname and procdate query task record
* @param procName
* @param procDate
* @return
*/
public
static
TaskRecordStatus
getTaskRecordState
(
String
procName
,
String
procDate
){
String
sql
=
String
.
format
(
"SELECT * FROM eamp_hive_log_hd WHERE PROC_NAME='%s' and PROC_DATE='%s'"
,
procName
,
procDate
);
List
<
TaskRecord
>
taskRecordList
=
getQueryResult
(
sql
);
// contains no record and sql exception
if
(
CollectionUtils
.
isEmpty
(
taskRecordList
)){
// exception
return
TaskRecordStatus
.
EXCEPTION
;
}
else
if
(
taskRecordList
.
size
()
>
1
){
return
TaskRecordStatus
.
EXCEPTION
;
}
else
{
TaskRecord
taskRecord
=
taskRecordList
.
get
(
0
);
if
(
taskRecord
==
null
){
return
TaskRecordStatus
.
EXCEPTION
;
}
Long
targetRowCount
=
taskRecord
.
getTargetRowCount
();
if
(
targetRowCount
<=
0
){
return
TaskRecordStatus
.
FAILURE
;
}
else
{
return
TaskRecordStatus
.
SUCCESS
;
}
}
}
}
}
escheduler-server/src/main/java/cn/escheduler/server/worker/runner/TaskScheduleThread.java
浏览文件 @
5c1b3f0e
...
@@ -19,15 +19,18 @@ package cn.escheduler.server.worker.runner;
...
@@ -19,15 +19,18 @@ package cn.escheduler.server.worker.runner;
import
cn.escheduler.common.Constants
;
import
cn.escheduler.common.Constants
;
import
cn.escheduler.common.enums.ExecutionStatus
;
import
cn.escheduler.common.enums.ExecutionStatus
;
import
cn.escheduler.common.enums.TaskRecordStatus
;
import
cn.escheduler.common.enums.TaskType
;
import
cn.escheduler.common.enums.TaskType
;
import
cn.escheduler.common.model.TaskNode
;
import
cn.escheduler.common.model.TaskNode
;
import
cn.escheduler.common.process.Property
;
import
cn.escheduler.common.process.Property
;
import
cn.escheduler.common.task.AbstractParameters
;
import
cn.escheduler.common.task.AbstractParameters
;
import
cn.escheduler.common.task.TaskTimeoutParameter
;
import
cn.escheduler.common.task.TaskTimeoutParameter
;
import
cn.escheduler.common.utils.CommonUtils
;
import
cn.escheduler.common.utils.CommonUtils
;
import
cn.escheduler.common.utils.DateUtils
;
import
cn.escheduler.common.utils.HadoopUtils
;
import
cn.escheduler.common.utils.HadoopUtils
;
import
cn.escheduler.common.utils.TaskParametersUtils
;
import
cn.escheduler.common.utils.TaskParametersUtils
;
import
cn.escheduler.dao.ProcessDao
;
import
cn.escheduler.dao.ProcessDao
;
import
cn.escheduler.dao.TaskRecordDao
;
import
cn.escheduler.dao.model.ProcessInstance
;
import
cn.escheduler.dao.model.ProcessInstance
;
import
cn.escheduler.dao.model.TaskInstance
;
import
cn.escheduler.dao.model.TaskInstance
;
import
cn.escheduler.server.utils.LoggerUtils
;
import
cn.escheduler.server.utils.LoggerUtils
;
...
@@ -199,6 +202,23 @@ public class TaskScheduleThread implements Callable<Boolean> {
...
@@ -199,6 +202,23 @@ public class TaskScheduleThread implements Callable<Boolean> {
if
(
task
.
getExitStatusCode
()
==
Constants
.
EXIT_CODE_SUCCESS
){
if
(
task
.
getExitStatusCode
()
==
Constants
.
EXIT_CODE_SUCCESS
){
status
=
ExecutionStatus
.
SUCCESS
;
status
=
ExecutionStatus
.
SUCCESS
;
// task recor flat : if true , start up qianfan
if
(
TaskRecordDao
.
getTaskRecordFlag
()
&&
TaskType
.
typeIsNormalTask
(
taskInstance
.
getTaskType
())){
Date
scheduleTime
=
processInstance
.
getScheduleTime
();
if
(
scheduleTime
==
null
){
scheduleTime
=
processInstance
.
getStartTime
();
}
// process exec time : yyyyMMdd format
String
scheduleDate
=
DateUtils
.
format
(
scheduleTime
,
Constants
.
YYYYMMDD
);
TaskRecordStatus
taskRecordState
=
TaskRecordDao
.
getTaskRecordState
(
taskInstance
.
getName
(),
scheduleDate
);
logger
.
info
(
"task record status : {}"
,
taskRecordState
);
if
(
taskRecordState
==
TaskRecordStatus
.
FAILURE
){
status
=
ExecutionStatus
.
FAILURE
;
}
}
}
else
if
(
task
.
getExitStatusCode
()
==
Constants
.
EXIT_CODE_KILL
){
}
else
if
(
task
.
getExitStatusCode
()
==
Constants
.
EXIT_CODE_KILL
){
status
=
ExecutionStatus
.
KILL
;
status
=
ExecutionStatus
.
KILL
;
}
else
{
}
else
{
...
...
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录