海豚调度器 – gitweixin

海豚调度器 5月 30,2025

海豚调度器3.x版本解决创建文件夹时提示未指定当前登录用户的租户”的问题

在 海豚调度器 DolphinScheduler 3.x 中，出现提示：

“未指定当前登录用户的租户”

这是因为 当前登录用户没有绑定租户信息（Tenant），而 DolphinScheduler 在资源操作（如创建文件夹、提交任务等）时必须绑定租户，以便在后台进行资源隔离和权限控制。

一、问题成因分析

海豚调度器的核心概念之一是 租户（Tenant）。每个用户在执行任务时都必须以某个租户身份来进行资源调度。

在 DolphinScheduler 3.x 中：

每个用户需要 绑定一个默认租户。
如果没有绑定租户，进行资源操作（如资源中心上传、目录创建）时就会报错。

二、解决步骤

✅ 步骤1：登录系统，进入租户管理

使用 admin 用户登录 DolphinScheduler Web UI。
进入菜单：[安全中心] -> [租户管理]。
创建一个租户，例如：
- 租户编码：default_tenant
- 描述：默认租户

注意：租户编码实际对应 Linux 上执行任务的系统用户，需与服务器用户对应。

✅ 步骤2：为用户分配租户

进入菜单：[安全中心] -> [用户管理]
找到你的目标用户，点击编辑。
在编辑用户界面，将租户设置为刚刚创建的租户 default_tenant。
保存设置。

✅ 步骤3：重新登录或刷新页面

分配租户之后，用户需要重新登录或刷新浏览器，以使绑定关系生效。

✅ 步骤4：重新尝试资源中心操作

现在再尝试在资源中心创建文件夹、上传资源等操作，应不会再出现“未指定当前登录用户的租户”的提示。

作者 east

海豚调度器 9月 24,2024

海豚调度器如何设置上游的表生成数据成功才继续执行

在海豚调度器（Dolphin Scheduler）1.3.5 中，你可以通过以下步骤设置工作流中的任务，以检查 Hive 表是否为空，并根据结果决定是否执行后续任务。以下是详细步骤：

步骤 1: 创建 Hive 表检查任务

创建一个 Shell 脚本任务：该任务将执行 Hive 查询以检查指定表是否为空。
- 在 Dolphin Scheduler 中，创建一个新的任务，选择 Shell 作为任务类型。
- 编写脚本如下：

#!/bin/bash
# 检查 Hive 表是否为空
HIVE_TABLE_NAME="your_hive_table_name"  # 替换为你的 Hive 表名

COUNT=$(hive -S -e "SELECT COUNT(*) FROM ${HIVE_TABLE_NAME};")

if [ "$COUNT" -eq 0 ]; then
    echo "Table is empty"
    exit 1  # 返回非零值表示表为空
else
    echo "Table is not empty"
    exit 0  # 返回零值表示表不为空
fi

保存并命名该任务，例如命名为 check_hive_table_empty。

步骤 2: 创建后续任务

创建后续的任务：这些任务将根据 Hive 表是否为空而决定是否执行。例如，创建一个新的任务，命名为 run_after_check。
设置相应的任务逻辑。这些可以是任何你需要执行的操作，比如数据处理、报告生成等。

步骤 3: 设置任务依赖关系

创建一个工作流：在 Dolphin Scheduler 中创建一个新的工作流。
添加任务：
- 将 check_hive_table_empty 任务添加到工作流中。
- 将 run_after_check 任务也添加到工作流中。
设置依赖关系：
- 选择 run_after_check 任务，设置其依赖于 check_hive_table_empty 任务。
- 在依赖配置中，选择“成功”状态。这意味着只有当 check_hive_table_empty 任务成功运行且表不为空时，run_after_check 才会被执行。

步骤 4: 测试和验证工作流

启动工作流：手动启动工作流，观察任务的执行情况。
查看日志：检查 check_hive_table_empty 的日志，确认其正确判断了 Hive 表是否为空。
验证后续任务：根据 Hive 表的状态，确认后续任务是否被正确执行或跳过。

注意事项

确保 Hive 和 Dolphin Scheduler 的连接正常。
检查权限设置，确保你有权限访问 Hive 表。
可以根据需要调整任务超时时间和重试策略，以确保工作流的稳定性。

通过以上步骤，你能够在 Dolphin Scheduler 中实现根据 Hive 表内容动态控制工作流的执行。

作者 east

mysql, 海豚调度器 9月 18,2024

解决Sqoop从mysql导出数据到hive结果为空时的报错

在海豚调度器1.3.5，用Sqoop从hive导入数据到mysql，有时由于计算结果为空，导致hive的表当天分区的数据为空，Sqoop导出到mysql时报错。海豚调度器设置了失败继续策略也没用，导致后面的工作流没办法继续执行下去。

可以在执行sqoop之前增加判断当天分区的数量是否为空，如果为空就不执行导出。

# 检查分区是否为空
empty=$(hive -S -e "SELECT COUNT(*) FROM cnsaas.ads_bigdata_iot_data WHERE dt='$yesday_date';" | awk '{print $1}')

if [ "$empty" -eq 0 ]; then
  echo "分区为空，不执行 Sqoop 导出。"
else
  echo "分区非空，执行 Sqoop 导出。"

完整的sqoop脚本如下：

#!/bin/bash

# 定义变量
host="your_host"
user="your_user"
pwd="your_password"
yesday_date=$(date -d "yesterday" +%Y-%m-%d)

# 检查分区是否为空
empty=$(hive -S -e "SELECT COUNT(*) FROM cnsaas.ads_bigdata_iot_data WHERE dt='$yesday_date';" | awk '{print $1}')

if [ "$empty" -eq 0 ]; then
  echo "分区为空，不执行 Sqoop 导出。"
else
  echo "分区非空，执行 Sqoop 导出。"
  sqoop export \
    --connect jdbc:mysql://$host:3306/zgcn?characterEncoding=UTF-8 \
    --username $user \
    --password $pwd \
    -m 1 \
    --table ads_bigdata_iot_data\
    --columns stat_date,data_type,stat_type,cu_pid_system_code,max_value,min_value,avg_value \
    --fields-terminated-by '\001' \
    --update-key stat_date,data_type,stat_type,cu_pid_system_code \
    --update-mode allowinsert \
    --input-null-string '\\N' \
    --input-null-non-string '\\N' \
    --null-string '\\N' --null-non-string '\\N' \
    --export-dir /user/hive/warehouse/cnsaas.db/ads_bigdata_iot_data/dt=$yesday_date/*
fi

作者 east

海豚调度器 6月 24,2024

用海豚调度器api启动工作流错误排查

在海豚调度器用api启动工作流，在海豚调度器的工作流实例看到是启动失败，但在任务失例又没看到。看启动工作流的代码觉得好像没问题，一时不得其解。

后来找到海豚调度器的日志dolphinscheduler-master.log，看到日志如下：

[ERROR] 2024-06-24 12:27:24.466 org.apache.dolphinscheduler.dao.utils.DagHelper:[101] - start node name [ads_bigdata_xxxx] is not in task node list [[TaskNode{id='tasks-71533', name='ads_bigdata_battery_warning_detail.sh', desc='null', type='SHELL', runFlag='NORMAL', loc='null', maxRetryTimes=0, retryInterval=1, params='{"rawScript":"sh cnsaas/ads_bigdata_battery_warning_detail.sh ${complement_date}","localParams":[],"resourceList":[{"res":"cnsaas/ads_bigdata_xxxx.sh","name":"ads_bigdata_xxxx.sh","id":6},{"res":"cnsaas/ads_bigdata_xxxx.sql","name":"ads_bigdata_xxxx.sql","id":102}]}', preTasks='[]', extras='null', depList=[], dependence='{}', taskInstancePriority=MEDIUM, timeout='{"enable":false,"strategy":""}', workerGroup='default'}]] 
[ERROR] 2024-06-24 12:27:24.466 org.apache.dolphinscheduler.server.master.runner.MasterExecThread:[347] - processDag is null
[INFO] 2024-06-24 12:27:24.468 org.apache.dolphinscheduler.server.master.runner.MasterExecThread:[315] - prepare process :10712 end
[ERROR] 2024-06-24 12:27:24.468 org.apache.dolphinscheduler.server.master.runner.MasterExecThread:[184] - master exec thread exception
java.lang.NullPointerException: null
	at org.apache.dolphinscheduler.dao.utils.DagHelper.parsePostNodes(DagHelper.java:284)
	at org.apache.dolphinscheduler.server.master.runner.MasterExecThread.submitPostNode(MasterExecThread.java:482)
	at org.apache.dolphinscheduler.server.master.runner.MasterExecThread.runProcess(MasterExecThread.java:832)
	at org.apache.dolphinscheduler.server.master.runner.MasterExecThread.executeProcess(MasterExecThread.java:200)
	at org.apache.dolphinscheduler.server.master.runner.MasterExecThread.run(MasterExecThread.java:181)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
[ERROR] 2024-06-24 12:27:24.468 org.apache.dolphinscheduler.server.master.runner.MasterExecThread:[185] - process execute failed, process id:10712

原来启动节点名称（ads_bigdata_xxx）不在任务节点列表中。这意味着在DAG定义中，可能没有包含这个节点，或者节点的名称有误。认真检查一下，果然是启动工作流时，任务节点的名称写得有问题，重新修改后果然正常启动工作流了。

作者 east

海豚调度器 6月 14,2024

海豚调度器执行impla工作流成功但没跑出数据原因和解决方案

在海豚调度器(DolphinScheduler)中执行Impala离线脚本时出现执行状态成功但实际上未产出数据，之后重跑又能成功的情况，可能是由多种因素引起的。下面是一些可能的原因及对应的解决方案：

可能原因及解决方案

资源限制或并发冲突
- 原因：在执行脚本时，如果系统资源（如CPU、内存或I/O）紧张，或者与其他任务并发执行时产生资源竞争，可能导致Impala查询或作业执行虽成功但因资源不足未正确完成。
- 解决方案：监控资源使用情况，适当调整任务的执行时间窗口以避免高峰时段，增加资源限制或优化资源分配策略。
Impala缓存问题
- 原因：Impala使用缓存来加速查询，有时候缓存中的数据可能已过期或不完整，导致查询结果不准确。
- 解决方案：在执行重要查询前，考虑清空或刷新Impala的查询缓存，或在查询语句中使用INVALIDATE METADATA命令来强制重新加载表的元数据。
数据一致性问题
- 原因：如果数据在查询执行过程中被其他任务修改（尤其是在分布式环境下），可能导致查询结果不一致。
- 解决方案：确保数据操作的原子性和一致性，使用事务控制，或在查询时锁定表（如果适用）以避免数据并发修改问题。
Impala版本或配置问题
- 原因：特定的Impala版本或配置设置可能导致某些查询行为不符合预期。
- 解决方案：检查Impala的版本和配置，确保使用的是稳定且支持当前查询特性的版本，并根据官方文档或社区反馈调整相关配置。
网络或连接问题
- 原因：瞬时的网络波动或连接问题可能导致查询虽然提交成功，但在数据传输阶段出现问题。
- 解决方案：检查网络连接稳定性，确保Impala服务端和客户端之间的通信没有问题。优化网络配置，如增加超时时间，使用更稳定的网络连接方式。
脚本逻辑问题
- 原因：离线脚本自身可能存在逻辑缺陷，比如依赖的数据源在某些条件下为空，导致查询看似成功但实际上没有返回数据。
- 解决方案：审查脚本逻辑，增加必要的错误处理和日志记录，确保脚本在各种数据情况下都能正确执行。进行单元测试或集成测试以验证脚本逻辑的健壮性。

综合建议

增加日志记录：在Impala脚本和DolphinScheduler任务中增加详细的日志记录，以便于问题发生时追踪问题根源。
监控与报警：实施全面的系统和任务执行监控，设置合理的报警机制，一旦发现异常及时介入。
定期维护与检查：定期检查系统配置、资源使用情况和软件版本，进行必要的维护和升级。

综合解决方案可参照：
海豚调度器自动监测每日报表及自动重跑异常工作流（综合应用可用代码

作者 east

海豚调度器 6月 14,2024

海豚调度器(DolphinScheduler)修改时区为东八区

海豚调度器设置了定时，执行的时间和设置时间不同，后来排查发现是时区问题。可以用下面方法和步骤来修改：

修改DolphinScheduler服务器时区

登录服务器：首先，通过SSH或其他方式登录到运行DolphinScheduler服务的服务器上。
查看当前时区：执行以下命令查看服务器当前的时区设置：Bash1timedatectl或Bash1date
修改时区：如果需要修改，可以使用以下命令将时区设置为您所需的时区。例如，要设置为上海时区（Asia/Shanghai），执行：sudo timedatectl set-timezone Asia/Shanghai或对于较旧的系统，可能需要使用：sudo ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime
重启服务：修改时区后，可能需要重启DolphinScheduler以及相关依赖服务（如MySQL）以确保新时区生效。

修改DolphinScheduler界面配置时区

登录DolphinScheduler Web UI：使用管理员账号登录DolphinScheduler的Web界面。
访问系统设置：导航至系统管理 > 系统设置。
修改时区：在系统设置页面中，找到“时区设置”部分，从下拉菜单中选择正确的时区。确保这个时区设置与您之前在服务器上设置的时区相匹配。
保存设置：修改完成后，记得点击“保存”按钮使更改生效。

如果是一些老版本的海豚调度器，是没有上面的设置的，需要修改配置文件：

找到application.properties配置文件：
- 在海豚调度器的部署目录中，找到conf文件夹。
- 在conf文件夹中，找到名为application.properties的配置文件。
编辑application.properties文件：
- 使用文本编辑器打开application.properties文件。
- 在文件中找到以下行：spring.jackson.time-zone=Asia/Shanghai这一行表示默认的时区设置为亚洲/上海时区。根据您的需求，将其更改为所需的时区。例如，如果您想使用美国东部标准时间，可以将其更改为：spring.jackson.time-zone=America/New_York
- 保存并关闭文件。
重启海豚调度器服务：
- 为了使时区设置立即生效，您需要重启海豚调度器的服务。
- 根据您的部署方式（如Docker、Kubernetes或手动部署），执行相应的命令来重启海豚调度器服务。
验证时区设置：
- 重启服务后，重新登录到海豚调度器管理界面。
- 创建一个新的定时任务，并设置一个未来的执行时间。
- 观察并记录任务实际执行的时间，以确保时区设置已正确应用。

作者 east

Java, 海豚调度器 6月 7,2024

海豚调度器调用api接口启动工作流（java版本实现）

海豚调度器调用api接口启动工作流（亲试可用），详细介绍怎样用python代码启动工作流，不过后来有的生成环境是安装在docker，不通外网，python环境不支持requests。

方案1：离线安装requests

方案2：改成用java语言现实，所有依赖包打包成jar。

import java.net.URI;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.http.HttpRequest;
import org.apache.http.NameValuePair;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.message.BasicNameValuePair;
import org.apache.http.util.EntityUtils;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.client.utils.URIBuilder;
import org.json.JSONArray;
import org.json.JSONObject;

public static void startWorkflow(String token, String projectName, String processDefinitionName, String processDefinitionId, String startNode) {
        // 构建请求URL和参数
        String url = DOLPHIN_SCHEDULER_URL + "/projects/" + projectName + "/executors/start-process-instance";

        List<NameValuePair> params = new ArrayList<>();
        params.add(new BasicNameValuePair("processDefinitionName", processDefinitionName));
        params.add(new BasicNameValuePair("processDefinitionId", processDefinitionId));
        params.add(new BasicNameValuePair("failureStrategy", "CONTINUE"));
        params.add(new BasicNameValuePair("warningType", "NONE"));
        params.add(new BasicNameValuePair("warningGroupId", "0"));
        params.add(new BasicNameValuePair("scheduleTime", ""));
        params.add(new BasicNameValuePair("runMode", "RUN_MODE_SERIAL"));
        params.add(new BasicNameValuePair("processInstancePriority", "MEDIUM"));
        params.add(new BasicNameValuePair("workerGroup", "default"));
        params.add(new BasicNameValuePair("timeout", "100"));
        params.add(new BasicNameValuePair("startNodeList", startNode));
        params.add(new BasicNameValuePair("taskDependType","TASK_ONLY" ));


        CloseableHttpClient client = null;
        try {
        URI uri = new URIBuilder(url)
                .addParameters(params)
                .build();

        client = HttpClients.createDefault();
        HttpPost httpPost = new HttpPost(uri);
        httpPost.setHeader("Content-Type", "application/json");
        httpPost.setHeader("token", token);


            CloseableHttpResponse response = client.execute(httpPost);
            HttpEntity entity = response.getEntity();
            String responseString = EntityUtils.toString(entity, "UTF-8");
            if (response.getStatusLine().getStatusCode() == 200) {
                System.out.println("Workflow started successfully: " + responseString);
            } else {
                System.out.println("Failed to start workflow: " + response.getStatusLine().getStatusCode());
            }
        } catch (Exception e) {
            System.out.println("Error starting workflow: " + e.getMessage());
        } finally {
            try {
                client.close();
            } catch (Exception e) {
                System.out.println("Error closing HttpClient: " + e.getMessage());
            }
        }
    }

作者 east

python, 海豚调度器 6月 7,2024

海豚调度器调用api接口来获取工作流信息（获取processDefinitionId）

在前面一文，海豚调度器调用api接口启动工作流（亲试可用），详细介绍processDefinitionId通过t_ds_process_definition来获取，并没有详细介绍如何用api调用。下面详细介绍如何用api获取。

获取工作流的信息：

#查询流程定义通过项目ID
def queryProcessDefinitionAllByProjectId(token,project_name, project_id):
    url = f"{dolphin_scheduler_base_url}/projects/{project_name}/process/queryProcessDefinitionAllByProjectId"
    params = {
        "projectId": project_id
    }
    headers = {
        "Content-Type": "application/json",
        "token": token
    }
    response = requests.get(url, headers=headers, params=params)

    if response.status_code == 200:
        return response.json()
    else:
        return None

解析工作流的信息，获取工作流名称和processDefinitionId的字典：

def extract_name_id_mapping(json_data):
    name_id_mapping = {}
    data_list = json_data.get('data', [])
    for item in data_list:
        name = item.get('name')
        id = item.get('id')
        if name and id:
            name_id_mapping[name] = id
    return name_id_mapping

调用如下：

 json_data = queryProcessDefinitionAllByProjectId(token,project_name, project_id)
        name_id_mapping = extract_name_id_mapping(json_data)
        print(name_id_mapping)

作者 east

海豚调度器 5月 8,2024

海豚调度器早期版本如何新增worker分组

在DolphinScheduler 1.3.5版本中，Worker分组通常是在部署时通过配置文件进行定义的，而不是在用户界面上直接操作。以下是在DolphinScheduler中新增Worker分组的一般步骤：

修改配置文件： DolphinScheduler的Worker分组信息通常在/conf目录下的配置文件中定义。你需要找到相关的配置文件，比如dolphinscheduler-env.sh，并添加或修改Worker分组的配置。
定义Worker分组：在配置文件中，你可以定义Worker的分组信息。例如，你可以添加一个新的Worker分组名为group1，然后在相应的配置项中指定这个分组。在海豚调度器是在config文件夹的install_config.conf来修改的，在下在这个配置项后面增加：workers=”cdh01:default,cdh02:default,cdh03:default”
配置Worker节点：接下来，你需要在集群中的Worker节点上进行配置，以确保它们能够注册到刚才创建的分组中。这通常涉及到修改每个Worker节点上的DolphinScheduler配置文件，指定它们属于哪个Worker分组。
重启Worker服务：修改配置文件后，需要重启所有Worker节点上的DolphinScheduler Worker服务，以使配置生效。
验证Worker分组： Worker服务重启后，你可以通过DolphinScheduler的Master服务日志来验证Worker节点是否成功注册到了新的分组中。
前端展示： DolphinScheduler的Web界面会自动展示配置文件中定义的Worker分组信息。如果Worker节点注册成功，你应该能够在界面上看到新的Worker分组及其状态。

请注意，具体的配置文件和参数可能会根据DolphinScheduler的不同版本而有所变化。如果你不确定如何操作，可以查阅DolphinScheduler的官方文档或在社区寻求帮助。

作者 east

分类归档海豚调度器