DolphinScheduler 3.1.8 跨任务参数传递实战Shell/SQL/Python 数据流转全解析1. 参数传递机制核心原理在DolphinScheduler中参数传递是构建复杂工作流的关键能力。不同于简单的任务编排跨任务数据流转需要理解三大核心机制作用域层级体系全局参数工作流级别可见通过${变量名}引用局部参数任务级别有效需显式声明OUT方向才能传递内置参数系统预置的时间变量如${system.biz.date}参数传递方向控制# Python任务输出参数示例 print(${setValue(output_param%s)} % result_value)生命周期管理参数生成上游任务setValue持久化存储数据库记录下游引用通过${参数名}获取关键提示参数传递本质是通过数据库持久化实现的中间状态共享而非内存直接传递2. Shell任务参数传递实战2.1 基础参数输出#!/bin/bash # 直接输出固定值 echo ${setValue(api_version1.0)} # 动态计算值传递 timestamp$(date %s) echo ${setValue(exec_timestamp$timestamp)}2.2 环境变量集成#!/bin/bash # 获取环境变量并传递 cluster_node${HOSTNAME} echo ${setValue(node_name${cluster_node})} # 命令结果传递 file_count$(ls | wc -l) echo ${setValue(total_files${file_count})}2.3 多参数批量传递#!/bin/bash # 生成JSON格式数据 stats_json{success: 200, failed: 3} # 拆解JSON传递多个参数 echo ${setValue(success_count$(echo $stats_json | jq .success))} echo ${setValue(fail_count$(echo $stats_json | jq .failed))}参数配置示例参数名方向数据类型示例值api_versionOUTVARCHAR1.0exec_timestampOUTLONG1630000000node_nameOUTVARCHARnode-013. SQL任务参数传递方案3.1 查询结果传递-- 单行结果传递字段名作为参数名 SELECT user_id AS id, user_name AS name FROM t_users WHERE id 1001; -- 多行结果自动转为JSON数组 SELECT * FROM t_order WHERE create_date ${system.biz.date};3.2 存储过程参数处理-- MySQL存储过程示例 DELIMITER // CREATE PROCEDURE sp_get_stats(OUT total_users INT, OUT active_rate DECIMAL(5,2)) BEGIN SELECT COUNT(*) INTO total_users FROM t_users; SELECT ROUND(SUM(is_active)/COUNT(*),2) INTO active_rate FROM t_users; END // DELIMITER ; -- 调用时声明OUT参数 CALL sp_get_stats(${setValue(total_users)}, ${setValue(active_rate)});3.3 动态SQL构建技巧-- 使用参数化查询 SELECT * FROM t_orders WHERE create_date BETWEEN ${start_date} AND ${end_date} LIMIT ${page_size} OFFSET ${page_num * page_size}; -- 表名动态化需注意SQL注入风险 SET table_name CONCAT(t_orders_, YEAR(CURDATE())); SET sql CONCAT(SELECT COUNT(*) FROM , table_name); PREPARE stmt FROM sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;4. Python任务高级参数处理4.1 结构化数据传递import json from datetime import datetime # 复杂对象序列化传递 analysis_result { model_version: v2.1, accuracy: 0.92, features: [age, income], timestamp: datetime.now().isoformat() } print(f${{setValue(analysis_result{json.dumps(analysis_result)})}}) # 多参数批量输出 params { batch_id: BATCH_2023, processed: 1500, skipped: 42 } for k, v in params.items(): print(f${{setValue({k}{v})}})4.2 机器学习场景集成import pandas as pd from sklearn.ensemble import RandomForestClassifier # 训练模型并传递关键参数 df pd.read_csv(data.csv) model RandomForestClassifier(n_estimators100) model.fit(df[[feature1, feature2]], df[label]) # 传递模型特征重要性 importances dict(zip(df.columns, model.feature_importances_)) print(f${{setValue(feature_importance{json.dumps(importances)})}}) # 传递评估指标 print(f${{setValue(model_score{model.score(X_test, y_test):.4f})}})4.3 异常处理规范try: # 业务逻辑处理 result process_data() print(f${{setValue(process_result{result})}}) except Exception as e: # 错误信息传递 print(f${{setValue(error_msg{str(e)})}}) # 状态码传递 print(${setValue(exit_code1)}) raise5. 跨类型任务串联实战5.1 完整工作流示例场景每日用户行为分析流水线Shell任务数据准备#!/bin/bash # 检查HDFS目录是否存在 hdfs dfs -test -d /data/raw/${system.biz.date} if [ $? -ne 0 ]; then echo ${setValue(need_init1)} else echo ${setValue(need_init0)} fiSQL任务数据聚合-- 依赖need_init参数 SELECT COUNT(DISTINCT user_id) AS uv, SUM(clicks) AS total_clicks FROM user_behavior WHERE dt ${system.biz.date} ${need_init 1 ? AND is_new 1 : }Python任务数据分析import pandas as pd from scipy import stats # 接收SQL结果 uv ${uv} clicks ${total_clicks} # 计算统计指标 ctr clicks / uv z_score stats.zscore([uv, global_avg_uv])[0] print(f${{setValue(ctr{ctr:.4f})}}) print(f${{setValue(z_score{z_score:.2f})}})5.2 参数调试技巧查看参数传递链路工作流实例 - 任务实例 - 参数列表日志搜索setValue关键字调试模式配置# 修改api-server配置 logging.level.org.apache.dolphinschedulerDEBUG常见问题排查表问题现象可能原因解决方案下游获取值为空参数方向未设置为OUT检查任务参数配置值被意外覆盖参数优先级冲突检查全局/局部参数命名特殊字符异常未正确转义使用json.dumps处理时间格式错误时区不一致统一使用UTC时间戳6. 性能优化与安全实践6.1 参数管理规范命名空间建议[模块]_[功能]_[变量类型] 示例 ads_user_uv_count dwd_order_amount_sum生命周期控制临时参数添加tmp_前缀敏感参数使用enc_前缀标记6.2 性能优化方案批量传递优化# 不推荐多次IO操作 for i in range(100): print(f${{setValue(param_{i}value)}}) # 推荐JSON打包传递 params {fparam_{i}: fvalue for i in range(100)} print(f${{setValue(batch_params{json.dumps(params)})}})类型转换对照表原始类型存储类型转换方式Python dictVARCHARjson.dumpsPandas DataFrameVARCHARto_json()Numpy arrayVARCHARjson.dumps6.3 安全防护措施敏感参数处理# 加密示例需提前配置密钥 encrypted$(echo secret_value | openssl enc -aes-256-cbc -md sha512 -a -pbkdf2 -iter 100000 -salt -pass pass:${ENCRYPT_KEY}) echo ${setValue(db_password${encrypted})}参数验证脚本import re def validate_param(key, value): # 名称规范检查 if not re.match(r^[a-z][a-z0-9_]{1,63}$, key): raise ValueError(fInvalid param name: {key}) # 值长度限制 if len(str(value)) 4096: raise ValueError(Parameter value too long) # 敏感词过滤 sensitive_words [password, secret, token] if any(word in key.lower() for word in sensitive_words): raise ValueError(Potential sensitive parameter detected)
DolphinScheduler 3.1.8 跨任务参数传递实战Shell/SQL/Python 数据流转全解析1. 参数传递机制核心原理在DolphinScheduler中参数传递是构建复杂工作流的关键能力。不同于简单的任务编排跨任务数据流转需要理解三大核心机制作用域层级体系全局参数工作流级别可见通过${变量名}引用局部参数任务级别有效需显式声明OUT方向才能传递内置参数系统预置的时间变量如${system.biz.date}参数传递方向控制# Python任务输出参数示例 print(${setValue(output_param%s)} % result_value)生命周期管理参数生成上游任务setValue持久化存储数据库记录下游引用通过${参数名}获取关键提示参数传递本质是通过数据库持久化实现的中间状态共享而非内存直接传递2. Shell任务参数传递实战2.1 基础参数输出#!/bin/bash # 直接输出固定值 echo ${setValue(api_version1.0)} # 动态计算值传递 timestamp$(date %s) echo ${setValue(exec_timestamp$timestamp)}2.2 环境变量集成#!/bin/bash # 获取环境变量并传递 cluster_node${HOSTNAME} echo ${setValue(node_name${cluster_node})} # 命令结果传递 file_count$(ls | wc -l) echo ${setValue(total_files${file_count})}2.3 多参数批量传递#!/bin/bash # 生成JSON格式数据 stats_json{success: 200, failed: 3} # 拆解JSON传递多个参数 echo ${setValue(success_count$(echo $stats_json | jq .success))} echo ${setValue(fail_count$(echo $stats_json | jq .failed))}参数配置示例参数名方向数据类型示例值api_versionOUTVARCHAR1.0exec_timestampOUTLONG1630000000node_nameOUTVARCHARnode-013. SQL任务参数传递方案3.1 查询结果传递-- 单行结果传递字段名作为参数名 SELECT user_id AS id, user_name AS name FROM t_users WHERE id 1001; -- 多行结果自动转为JSON数组 SELECT * FROM t_order WHERE create_date ${system.biz.date};3.2 存储过程参数处理-- MySQL存储过程示例 DELIMITER // CREATE PROCEDURE sp_get_stats(OUT total_users INT, OUT active_rate DECIMAL(5,2)) BEGIN SELECT COUNT(*) INTO total_users FROM t_users; SELECT ROUND(SUM(is_active)/COUNT(*),2) INTO active_rate FROM t_users; END // DELIMITER ; -- 调用时声明OUT参数 CALL sp_get_stats(${setValue(total_users)}, ${setValue(active_rate)});3.3 动态SQL构建技巧-- 使用参数化查询 SELECT * FROM t_orders WHERE create_date BETWEEN ${start_date} AND ${end_date} LIMIT ${page_size} OFFSET ${page_num * page_size}; -- 表名动态化需注意SQL注入风险 SET table_name CONCAT(t_orders_, YEAR(CURDATE())); SET sql CONCAT(SELECT COUNT(*) FROM , table_name); PREPARE stmt FROM sql; EXECUTE stmt; DEALLOCATE PREPARE stmt;4. Python任务高级参数处理4.1 结构化数据传递import json from datetime import datetime # 复杂对象序列化传递 analysis_result { model_version: v2.1, accuracy: 0.92, features: [age, income], timestamp: datetime.now().isoformat() } print(f${{setValue(analysis_result{json.dumps(analysis_result)})}}) # 多参数批量输出 params { batch_id: BATCH_2023, processed: 1500, skipped: 42 } for k, v in params.items(): print(f${{setValue({k}{v})}})4.2 机器学习场景集成import pandas as pd from sklearn.ensemble import RandomForestClassifier # 训练模型并传递关键参数 df pd.read_csv(data.csv) model RandomForestClassifier(n_estimators100) model.fit(df[[feature1, feature2]], df[label]) # 传递模型特征重要性 importances dict(zip(df.columns, model.feature_importances_)) print(f${{setValue(feature_importance{json.dumps(importances)})}}) # 传递评估指标 print(f${{setValue(model_score{model.score(X_test, y_test):.4f})}})4.3 异常处理规范try: # 业务逻辑处理 result process_data() print(f${{setValue(process_result{result})}}) except Exception as e: # 错误信息传递 print(f${{setValue(error_msg{str(e)})}}) # 状态码传递 print(${setValue(exit_code1)}) raise5. 跨类型任务串联实战5.1 完整工作流示例场景每日用户行为分析流水线Shell任务数据准备#!/bin/bash # 检查HDFS目录是否存在 hdfs dfs -test -d /data/raw/${system.biz.date} if [ $? -ne 0 ]; then echo ${setValue(need_init1)} else echo ${setValue(need_init0)} fiSQL任务数据聚合-- 依赖need_init参数 SELECT COUNT(DISTINCT user_id) AS uv, SUM(clicks) AS total_clicks FROM user_behavior WHERE dt ${system.biz.date} ${need_init 1 ? AND is_new 1 : }Python任务数据分析import pandas as pd from scipy import stats # 接收SQL结果 uv ${uv} clicks ${total_clicks} # 计算统计指标 ctr clicks / uv z_score stats.zscore([uv, global_avg_uv])[0] print(f${{setValue(ctr{ctr:.4f})}}) print(f${{setValue(z_score{z_score:.2f})}})5.2 参数调试技巧查看参数传递链路工作流实例 - 任务实例 - 参数列表日志搜索setValue关键字调试模式配置# 修改api-server配置 logging.level.org.apache.dolphinschedulerDEBUG常见问题排查表问题现象可能原因解决方案下游获取值为空参数方向未设置为OUT检查任务参数配置值被意外覆盖参数优先级冲突检查全局/局部参数命名特殊字符异常未正确转义使用json.dumps处理时间格式错误时区不一致统一使用UTC时间戳6. 性能优化与安全实践6.1 参数管理规范命名空间建议[模块]_[功能]_[变量类型] 示例 ads_user_uv_count dwd_order_amount_sum生命周期控制临时参数添加tmp_前缀敏感参数使用enc_前缀标记6.2 性能优化方案批量传递优化# 不推荐多次IO操作 for i in range(100): print(f${{setValue(param_{i}value)}}) # 推荐JSON打包传递 params {fparam_{i}: fvalue for i in range(100)} print(f${{setValue(batch_params{json.dumps(params)})}})类型转换对照表原始类型存储类型转换方式Python dictVARCHARjson.dumpsPandas DataFrameVARCHARto_json()Numpy arrayVARCHARjson.dumps6.3 安全防护措施敏感参数处理# 加密示例需提前配置密钥 encrypted$(echo secret_value | openssl enc -aes-256-cbc -md sha512 -a -pbkdf2 -iter 100000 -salt -pass pass:${ENCRYPT_KEY}) echo ${setValue(db_password${encrypted})}参数验证脚本import re def validate_param(key, value): # 名称规范检查 if not re.match(r^[a-z][a-z0-9_]{1,63}$, key): raise ValueError(fInvalid param name: {key}) # 值长度限制 if len(str(value)) 4096: raise ValueError(Parameter value too long) # 敏感词过滤 sensitive_words [password, secret, token] if any(word in key.lower() for word in sensitive_words): raise ValueError(Potential sensitive parameter detected)
