Are you satisfied with the way you currently manage the dependencies in your ETL? In Part 1 of this series, I talked about the features I expect from a dependency management system, and what are the main possibilities offered (directly or indirectly) by Data Services. In Part 2, I proposed an architecture (structure and expected behavior) for a dependency management system inside Data Services. Now I will give you the implementation details, while a feedback on how it went “in real life” as well as possible improvements will come in part 4. So how do we implement this theoretical solution in Data Services?
Empecemos con la creación de las tablas necesarias (la sintaxis a continuación es para Oracle, las cosas pueden ser ligeramente diferentes para su DBMS).
Crear tabla FLOW_DEPENDENCIES (
FLOW_NAME VARCHAR2 (100) no es nulo,
PREREQUISITE VARCHAR2(100)
not Null
,
Restricción FLOW_DEP_PK Clave principal (FLOW_NAME, PREREQUISITE) Habilitar);
Create Table
FLOW_STATUS (
JOB_KEY NUMBER(28, 0)
not Null
,
FLOW_NAME VARCHAR2(100)
not Null
,
STATUS VARCHAR2(50)
not Null
,
Restricción FLOW_STATUS_PK Clave principal (JOB_KEY, FLOW_NAME) Habilitar);
Ahora, ¿qué pasa con la parte de servicios de datos? Suponiendo que queremos actualizar un ETL existente, queremos que la solución práctica sea lo más rápida posible para implementar y permitir un fácil mantenimiento. Esto implica que los componentes deben ser tan estandarizados como sea posible, con una personalización mínima a cero para cada flujo.
Imaginemos el caso en el que tenemos 3 flujos (Flujos de trabajo WF_A y WF_C, Flujo de datos DF_B) que realizan una serie de operaciones, por ejemplo la actualización de la dimensión de la cuenta. Tienen los mismos prerrequisitos y pueden ser considerados como un flujo único (para las necesidades de nuestro sistema de gestión de dependencias).
Flujos de trabajo
Vamos a crear un flujo de trabajo para gestionar las dependencias de estos 3 flujos. La denominaremos WFDM_Account (para la Gestión de dependencias de flujo de trabajo - Dimensión de la cuenta).
Dependencias ETL
En el interior, pondremos un objeto condicional que ejecutará el flujo solo si es necesario.
Contenido de la cuenta WFDM
Antes de entrar en los detalles del guión, es hora de pensar en las variables que necesitaremos:
$ G_Job_Key es el identificador único para este trabajo específico. Suponemos que se ha generado anteriormente en el trabajo.
$ G_Previous_Job_Key es el identificador único de un trabajo anterior. Se establece manualmente, y sólo en caso de que queramos volver a iniciar el ETL y ejecutar sólo los flujos que falló o no se ejecutó. En el ejemplo de la parte 2 de este artículo, habríamos ejecutado el ETL por segunda vez estableciendo $ G_Previous_Job_Key = 1 (la clave de trabajo de la primera ejecución).
Para facilitar el script, utilizaré cuatro funciones personalizadas. Los nombres y el propósito están abajo, el código vendrá más tarde.
Check_Flow ($ P_Flow_Name) es la función principal. Devuelve "Sí" si se permite que el flujo se ejecute. De lo contrario devuelve "No" e inserta una fila con el motivo correspondiente en la tabla FLOW_STATUS.
Get_Flow_Status ($ P_Job_Key, $ P_Flow_Name) devuelve el estado de un flujo dado y una clave de trabajo determinada.
Insert_Flow_Status ($ P_Job_Key, $ P_Flow_Name, $ P_Status) inserta una nueva fila de estado para un flujo dado y una clave de trabajo determinada.
Is_Prerequisite_OK ($ P_Job_Key, $ P_Flow_Name) devuelve "No" si cualquier requisito previo de un flujo dado para una ejecución de trabajo determinada falta, "Sí" en caso contrario.
A continuación se muestra el código de la función Check_Flow ($ P_Flow_Name)
Si (($ G_Previous_Job_Key no es Null) y (Get_Flow_Status ($ G_Previous_Job_Key, $ P_Flow_Name) en ('Éxito', 'Ya se ejecutan'))))
Comenzar
Insert_Flow_Status ($ G_Job_Key, $ P_Flow_Name, 'Ya se ejecutan');
Devuelva 'No';
FIN
Más
Comenzar
Si (Is_Prerequisite_OK ($ G_Job_Key, $ P_Flow_Name) = 'No')
Comenzar
Insert_Flow_Status ($ G_Job_Key, $ P_Flow_Name, 'Faltante Prerrequisito');
Devuelva 'No';
FIN
Else devuelve 'sí';
FIN
Las dos funciones Get_Flow_Status y Insert_Flow_Status son fáciles:
Get_Flow_Status ($ P_Job_Key, $ P_Flow_Name):
Si ($ P_Job_Key no es nulo)
Return sql('Datawarehouse', 'select STATUS from FLOW_STATUS where JOB_KEY = [$P_Job_Key] and FLOW_NAME = ');
Más
Devuelve Nulo;
Insert_Flow_Status ($ P_Job_Key, $ P_Flow_Name, $ P_Status):
Sql('Datawarehouse', 'insert into FLOW_STATUS values ([$P_Job_Key], , ');
Devuelve 0;
La función Is_Prerequisite_OK ($ P_Job_Key, $ P_Flow_Name) es un poco más complicada debido a la combinación. Compara el número de filas de la tabla FLOW_STATUS que tienen un estado "Success" / "Already run" con el número de filas de la tabla FLOW_DEPENDENCIES.
Si
(Sql ('Datawarehouse', '
seleccione cuenta (*)
De FLOW_DEPENDENCIES D
Únete a FLOW_STATUS S
En D.PREREQUISITE = S.FLOW_NAME
Donde S.JOB_KEY = [$ P_Job_Key]
Y S.STATUS en ('Éxito', 'Ya funcionan')
and D.FLOW_NAME = ')
=
(Sql ('Datawarehouse', '
seleccione cuenta (*)
De FLOW_DEPENDENCIES
where FLOW_NAME = ')))
Devuelva 'Sí';
Else devuelve 'no';
Ahora podemos echar un vistazo al contenido del objeto condicional Run_flow.
La condición if comprueba si se permite que el flujo de trabajo actual se ejecute. Si ese no es el caso, no necesitamos hacer nada, por lo que la parte "Else" del objeto está vacía. Pero si se permite que el flujo de trabajo se ejecute, lo ejecutamos (aquí WF_A, DF_B y WF_C). Para actualizar la tabla FLOW_STATUS en caso de éxito, utilizamos un script "Success" que simplemente inserta una línea de estado:
Insert_Flow_Status ($ G_Job_Key, workflow_name (), 'Éxito');
Y para actualizar la tabla FLOW_STATUS en caso de fallo, utilizamos una estructura try-catch con un script simple en el objeto Catch:
Insert_Flow_Status ($ G_Job_Key, workflow_name (), 'Failure');
Estructura de try-catch
¡Eso es! Una vez que hayas probado eso, te recomiendo que reproduzcas el flujo padre, cambia el nombre a WFDM_Template. Ahora puede replicar la plantilla tantas veces como sea necesario y simplemente insertar los flujos de trabajo y los flujos de datos necesarios.
El último paso para realmente hacer que las dependencias funcionen es, por supuesto, para insertar todas las parejas de flujo / requisito previo en la tabla FLOW_DEPENDENCIES. Se basarán en los nombres de los flujos de trabajo replicados de WFDM_Template.
En la última parte de este artículo, le daré un comentario sobre cómo se implementó en un proyecto de cliente, así como las posibles mejoras. Hasta entonces, espero su opinión sobre esta solución. ¿Se ve bien? ¿Tienes otra solución? Deje un comentario a continuación.