Seguimiento del linaje de datos a escala: cómo esta plataforma fuera de línea maneja petabytes diariamente

Este artículo presenta una plataforma de gobierno de datos fuera de línea basada en Delfinschedulerabordando problemas como la dependencia de las tareas agujeros negros y cuellos de botella de escalabilidad. Al implementar la compilación dinámica basada en YAML y la captura de linaje automático, la plataforma permite una gestión eficiente de dependencia de tareas y seguimiento de datos. Aprovecha el Neo4J Base de datos de gráficos para el almacenamiento de linaje, admitiendo el análisis de impacto de segundo nivel y la localización de la causa raíz. Además, con una herramienta de importación de datos de alto rendimiento autodesarrollada, la eficiencia de transmisión de datos mejora significativamente.

Antecedentes y desafíos

Bajo la presión de procesar petabytes de datos diariamente, el sistema de programación authentic enfrentó dos problemas principales:

1. Dependencia de la tarea agujeros negros: Las dependencias de tareas del sistema cruzado (colmena/TIDB/Starrocks) se mantuvieron manualmente, lo que resultó en tiempos de solución de problemas superiores a los 30 minutos.
2. Cuellos de botella de escalabilidad: Un programador de un solo punto no pudo manejar miles de tareas concurrentes. La falta de un mecanismo de reintento condujo a tasas de latencia de datos que superan el 5%.

Pila de tecnología

Diseño de arquitectura central

Implementaciones técnicas clave:

1. Compilación dinámica de YAML

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

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

2. Captura de linaje automático

• Intercepta los planes de ejecución de SQL para analizar tablas de entrada/salida
• Para tareas que no son de SQL, usa ganchos para capturar rutas de archivos

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

Soluciones a desafíos clave

1. Plan de migración de incidentes cero

• Comparación de doble carrera: Ejecute sistemas antiguos y nuevos en paralelo; Use la herramienta Datadiff para verificar la consistencia de los resultados
• Lanzamiento canario: Break up Tráfico por unidad de negocio en etapas
• Mecanismo de reversión: Capacidad completa de reversión en 5 minutos

2. Herramienta de importación de alto rendimiento autodesarrollado

Optimizaciones clave:

• Envío por lotes usando GO’s Coroutine Pool
• Estrategia de ajuste de búfer dinámico

func (w *StarrocksWriter) batchCommit() {
    for {
        choose {
        case batch := <-w.batchChan:
            w.doBrokerLoad(batch) 
            // Dynamically regulate batch dimension
            w.adjustBatchSize(len(batch)) 
        }
    }
}

Implementación de gestión de linaje

Los datos de linaje se almacenan en la base de datos del gráfico Neo4J, habilitando:

• Análisis de impacto: Localice el alcance afectado de un cambio a nivel de tabla en segundos
• Análisis de causa raíz: Hint la fuente de un problema dentro de los 30 segundos durante las fallas
• Auditoría de cumplimiento: Cumple con los requisitos de trazabilidad de datos de GDPR

Ganancias de rendimiento