This article introduces an offline data governance platform built on DolphinScheduler, addressing issues like task dependency black holes and scalability bottlenecks. By implementing YAML-based dynamic compilation and automatic lineage capture, the platform enables efficient task dependency management and data tracking. It leverages the Neo4j graph database for lineage storage, supporting second-level impact analysis and root cause localization. Additionally, with a self-developed high-performance data import tool, data transmission efficiency is significantly improved.

Background and Challenges

Under the pressure of processing petabytes of data daily, the original scheduling system faced two major issues:

1. Task dependency black holes: Cross-system task dependencies (Hive/TiDB/StarRocks) were manually maintained, resulting in troubleshooting times exceeding 30 minutes.
2. Scalability bottlenecks: A single-point scheduler couldn't handle thousands of concurrent tasks. The lack of a retry mechanism led to data latency rates surpassing 5%.

Technology Stack

Core Architecture Design

Key Technical Implementations:

1. YAML Dynamic Compilation

type TaskDAG struct {
    Nodes []Node `yaml:"nodes"` 
    Edges []Edge `yaml:"edges"`
}

func GenerateWorkflow(yamlPath string) (*ds.WorkflowDefine, error) {
    data := os.ReadFile(yamlPath)
    var dag TaskDAG
    yaml.Unmarshal(data, &dag)
    // Convert to DolphinScheduler DAG structure
    return buildDSDAG(dag) 
}

2. Automatic Lineage Capture

• Intercepts SQL execution plans to parse input/output tables
• For non-SQL tasks, uses hooks to capture file paths

# StarRocks Broker Load Lineage Capture
def capture_brokerload(job_id):
    job = get_job_log(job_id)
    return {
      "input": job.params["hdfs_path"],
      "output": job.db_table 
    }

Solutions to Key Challenges

1. Zero-Incident Migration Plan

• Dual-run comparison: Run both old and new systems in parallel; use the DataDiff tool to verify result consistency
• Canary release: Split traffic by business unit in stages
• Rollback mechanism: Full rollback capability within 5 minutes

2. Self-Developed High-Performance Import Tool

Key Optimizations:

• Batch submission using Go's coroutine pool
• Dynamic buffer adjustment strategy

func (w *StarrocksWriter) batchCommit() {
    for {
        select {
        case batch := <-w.batchChan:
            w.doBrokerLoad(batch) 
            // Dynamically adjust batch size
            w.adjustBatchSize(len(batch)) 
        }
    }
}

Lineage Management Implementation

Lineage data is stored in the Neo4j graph database, enabling:

• Impact Analysis: Locate the affected scope of a table-level change within seconds
• Root Cause Analysis: Trace the source of an issue within 30 seconds during failures
• Compliance Auditing: Meets GDPR data traceability requirements

Performance Gains