DolphinScheduler 3.1.8 与 Flink/Spark 集成5个典型ETL工作流实战在大数据生态系统中任务调度系统与计算引擎的高效协同是构建稳定数据管道的核心。本文将深入探讨如何通过DolphinScheduler 3.1.8版本实现与Flink流处理和Spark批处理的无缝集成并通过五个典型场景展示实际应用方案。1. 环境配置与基础集成1.1 计算引擎环境准备在开始集成前需确保计算引擎环境满足以下条件Flink环境要求版本兼容性支持1.13版本集群模式Standalone/YARN/Kubernetes网络访问Worker节点需能访问Flink JobManager REST接口Spark环境要求版本支持Spark 2.4/3.x部署模式Cluster/Client模式资源管理与YARN或Kubernetes集成时需配置队列资源注意生产环境建议将引擎依赖包如flink-dist、spark-core预先部署在所有Worker节点的libs目录下路径通常为/opt/dolphinscheduler/libs1.2 DolphinScheduler服务配置修改common.properties关键参数# Flink配置 flink.home/opt/flink-1.15 flink.app.status.poll.interval10s # Spark配置 spark.home/opt/spark-3.2 spark.masteryarn spark.deploy.modecluster通过API创建计算引擎集群配置# 创建Spark集群配置示例 curl -X POST \ http://ds-server:12345/dolphinscheduler/clusters/create \ -H Token: YOUR_ACCESS_TOKEN \ -d { name: spark_prod, config: spark.masteryarn\nspark.executor.memory4g, type: SPARK }2. 实时数据入湖工作流设计2.1 Kafka到HDFS的实时ETL该工作流实现从Kafka消费数据经Flink实时处理写入HDFS的完整流程。DAG节点组成Kafka源配置节点定义消费的Topic和反序列化方式Flink SQL处理节点执行数据清洗和转换HDFS Sink节点配置写入路径和文件滚动策略关键Flink任务参数示例-- Flink SQL节点配置 CREATE TABLE kafka_source ( user_id STRING, event_time TIMESTAMP(3), metadata ROWip STRING, device STRING ) WITH ( connector kafka, topic user_events, properties.bootstrap.servers kafka:9092, format json ); CREATE TABLE hdfs_sink ( dt STRING, hour STRING, user_count BIGINT ) PARTITIONED BY (dt, hour) WITH ( connector filesystem, path hdfs://cluster/data/user_stats, format parquet, sink.rolling-policy.file-size 128MB ); INSERT INTO hdfs_sink SELECT DATE_FORMAT(event_time, yyyy-MM-dd) AS dt, DATE_FORMAT(event_time, HH) AS hour, COUNT(DISTINCT user_id) AS user_count FROM kafka_source GROUP BY TUMBLE(event_time, INTERVAL 1 HOUR), DATE_FORMAT(event_time, yyyy-MM-dd), DATE_FORMAT(event_time, HH)2.2 工作流参数化设计通过全局参数实现动态配置参数名示例值描述kafka.brokerskafka:9092Kafka集群地址sink.path/data/${bizDate}HDFS写入路径checkpoint.interval30000Flink检查点间隔(ms)3. 离线批处理工作流优化3.1 多阶段Spark SQL处理典型数仓分层处理流程包含ODS→DWD→DWS三层转换graph TD A[ODS层数据加载] --|Spark SQL| B[DWD层维度关联] B --|Spark SQL| C[DWS层聚合计算] C --|Spark SQL| D[ADS层结果导出]关键配置技巧资源动态分配根据数据量调整Executor数量spark-submit \ --conf spark.dynamicAllocation.enabledtrue \ --conf spark.shuffle.service.enabledtrue \ --conf spark.dynamicAllocation.maxExecutors20分区策略优化对DWD层表按日期分区-- 动态分区设置 SET hive.exec.dynamic.partitiontrue; SET hive.exec.dynamic.partition.modenonstrict; INSERT OVERWRITE TABLE dwd.user_behavior PARTITION(dt${bizDate}) SELECT user_id, item_id, behavior_type, from_unixtime(event_time) AS action_time FROM ods.kafka_user_events WHERE dt${bizDate};3.2 依赖管理实践通过DolphinScheduler的Dependent节点实现跨工作流依赖上游工作流结束时写入标记文件# 在Shell节点中执行 hdfs dfs -touchz /checkpoints/${workflowInstanceId}.success下游工作流通过条件依赖检查# dependent节点Python脚本 import subprocess result subprocess.run( [hdfs, dfs, -test, -e, f/checkpoints/{upstream_instance}.success], stdoutsubprocess.PIPE ) exit(0 if result.returncode 0 else 1)4. 混合计算场景实现4.1 FlinkSpark混合管道实时维度表更新结合离线计算的典型案例工作流结构Flink实时任务监听MySQL binlog更新Redis维度数据Spark离线任务每日全量刷新HBase维度表数据一致性检查对比Redis与HBase关键指标Flink CDC配置示例// 构建MySQL CDC源 DebeziumSourceFunctionString sourceFunction MySQLSource.Stringbuilder() .hostname(mysql-host) .port(3306) .databaseList(inventory) .tableList(inventory.products) .username(flinkuser) .password(password) .deserializer(new JsonDebeziumDeserializationSchema()) .build(); // 写入Redis的Sink实现 RedisSinkString redisSink new RedisSink( new FlinkJedisPoolConfig.Builder().setHost(redis).build(), new RedisProductUpdateMapper() );4.2 资源隔离方案通过多租户实现计算资源隔离YARN队列配置!-- capacity-scheduler.xml -- queue nameds_flink minResources10000 mb,10 vcores/minResources maxResources50000 mb,50 vcores/maxResources /queue queue nameds_spark minResources20000 mb,20 vcores/minResources maxResources100000 mb,100 vcores/maxResources /queueDolphinScheduler租户绑定-- 在数据库中添加租户队列映射 INSERT INTO t_ds_tenant_queue_relation (tenant_id, queue_name) VALUES (1, ds_flink), (2, ds_spark);5. 生产环境最佳实践5.1 高可用保障措施Master-Worker架构优化组件配置项推荐值说明Mastermaster.exec.threadsCPU核心数×2控制并行工作流数Workerworker.exec.threadsCPU核心数×1.5控制并行任务数所有节点heartbeat.interval10s心跳检测间隔ZooKeeper关键配置# zoo.cfg tickTime2000 initLimit10 syncLimit5 maxClientCnxns100 minSessionTimeout4000 maxSessionTimeout400005.2 监控与告警集成Prometheus监控指标采集配置DolphinScheduler暴露指标# application.yaml metrics: enabled: true exporter: type: prometheus port: 12346关键告警规则示例# prometheus_rules.yml - alert: MasterTaskQueueFull expr: ds_master_task_queue_size 1000 for: 5m labels: severity: critical annotations: summary: Master任务队列积压 (instance {{ $labels.instance }}) description: 任务队列持续高位当前值: {{ $value }}邮件告警模板配置{ type: EMAIL, name: prod_alert, params: { receivers: data-teamcompany.com, serverHost: smtp.exmail.qq.com, serverPort: 465, sender: ds-alertcompany.com, enableSmtpAuth: true, user: ds-alert, password: xxxxxx, starttlsEnable: false, sslEnable: true } }6. 性能调优实战6.1 大规模工作流优化千万级任务调度优化方案数据库优化-- 建立关键表索引 CREATE INDEX idx_command_create_time ON t_ds_command(create_time); CREATE INDEX idx_process_instance ON t_ds_process_instance(id, state); -- 历史数据归档策略 DELETE FROM t_ds_task_instance WHERE end_time DATE_SUB(NOW(), INTERVAL 30 DAY);工作流拆分原则单工作流任务数不超过200个复杂DAG拆分为子工作流使用子流程节点进行嵌套调用6.2 计算引擎参数调优Flink作业典型配置# flink-conf.yaml taskmanager.numberOfTaskSlots: 4 parallelism.default: 10 state.backend: rocksdb state.checkpoints.dir: hdfs://cluster/flink/checkpoints state.savepoints.dir: hdfs://cluster/flink/savepointsSpark性能关键参数spark-submit \ --conf spark.sql.adaptive.enabledtrue \ --conf spark.sql.shuffle.partitions200 \ --conf spark.executor.memoryOverhead1g \ --conf spark.memory.fraction0.7 \ --conf spark.serializerorg.apache.spark.serializer.KryoSerializer7. 异常处理与故障恢复7.1 失败策略配置DolphinScheduler提供多种失败处理机制策略类型适用场景配置方式继续执行独立子任务失败工作流定义→失败策略→继续结束流程关键路径失败工作流定义→失败策略→结束条件分支选择性重试使用条件分支节点自动重试配置示例{ taskInstance: { retryTimes: 3, retryInterval: 300, failureStrategy: CONTINUE } }7.2 数据一致性保障端到端精确一次方案FlinkKafka// 启用端到端精确一次 env.enableCheckpointing(60000); env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE); env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000); // Kafka生产者配置 properties.setProperty(enable.idempotence, true); properties.setProperty(transactional.id, flink-producer);SparkHDFS// 使用Delta Lake实现ACID df.write .format(delta) .mode(overwrite) .option(replaceWhere, sdt${partitionValue}) .save(/data/events)8. 扩展开发与API集成8.1 自定义任务类型开发开发SparkSQL任务类型的完整流程实现TaskPlugin接口public class SparkSQLTask extends AbstractTask { Override public AbstractParameters getParameters() { return new SparkSQLParameters(); } Override public void handle() throws Exception { String sql sparkSQLParameters.getSql(); String conf sparkSQLParameters.getSparkConfig(); // 构建SparkSession SparkSession spark SparkSession.builder() .config(parseConfig(conf)) .enableHiveSupport() .getOrCreate(); try { spark.sql(sql).show(); setExitStatusCode(Constants.EXIT_CODE_SUCCESS); } catch (Exception e) { logger.error(Execute SparkSQL failed, e); setExitStatusCode(Constants.EXIT_CODE_FAILURE); } } }注册插件到resources/plugin.propertiessparksqlorg.apache.dolphinscheduler.plugin.task.sparksql.SparkSQLTask8.2 REST API深度集成常用API操作示例创建工作流定义import requests url http://ds-server:12345/dolphinscheduler/projects/{projectName}/process-definition payload { name: flink_etl, description: 实时数据清洗流程, globalParams: {\bizDate\:\${system.datetime}\}, locations: [ { taskDefinition: { code: flink_task_1, type: FLINK, params: { mainClass: com.etl.FlinkJob, programArgs: --date ${bizDate} } }, x: 100, y: 200 } ] } response requests.post( url, jsonpayload, headers{Token: your_access_token} )补数操作APIcurl -X POST \ http://ds-server:12345/dolphinscheduler/projects/{projectName}/executors/execute \ -H Token: YOUR_TOKEN \ -d { processDefinitionCode: workflow_123, scheduleTime: 2026-01-01 00:00:00,2026-01-07 23:59:59, failureStrategy: CONTINUE, warningType: NONE, execType: REPEAT_RUNNING }