소규모 데이터 웨어하우스에서 ETL 프로세스 모니터링

많은 사람들이 특수 도구를 사용하여 데이터를 관계형 데이터베이스로 추출, 변환 및 로드하는 절차를 만듭니다. 작업 도구 프로세스가 기록되고 오류가 수정됩니다.

오류가 발생한 경우 로그에는 도구가 작업을 완료하지 못했고 어떤 모듈(종종 java)이 어디에서 중지되었는지에 대한 정보가 포함됩니다. 마지막 줄에서 데이터베이스 오류(예: 테이블 고유 키 위반)를 찾을 수 있습니다.

ETL 오류 정보가 어떤 역할을 하는지에 대한 질문에 답하기 위해 지난 XNUMX년 동안 발생한 모든 문제를 다소 큰 저장소에 분류했습니다.

데이터베이스 오류에는 공간 부족, 연결 끊김, 세션 중단 등이 포함됩니다.

논리적 오류에는 테이블 키 위반, 유효하지 않은 개체, 개체에 대한 액세스 부족 등이 포함됩니다.
스케줄러가 정시에 시작되지 않거나 멈출 수 있습니다.

간단한 실수는 수정하는 데 오래 걸리지 않습니다. 좋은 ETL은 대부분을 자체적으로 처리할 수 있습니다.

복잡한 버그로 인해 데이터 원본을 탐색하기 위해 데이터 작업을 위한 절차를 발견하고 테스트해야 합니다. 종종 변경 테스트 및 배포가 필요합니다.

따라서 모든 문제의 절반은 데이터베이스와 관련이 있습니다. 모든 실수의 48%는 단순한 실수입니다.
모든 문제의 XNUMX/XNUMX은 스토리지 논리 또는 모델 변경과 관련이 있으며 이러한 오류의 절반 이상이 복잡합니다.

모든 문제의 18/XNUMX 미만이 작업 스케줄러와 관련이 있으며 그 중 XNUMX%는 단순한 오류입니다.

일반적으로 발생하는 모든 오류의 22%는 복잡하며 이를 수정하는 데 가장 많은 주의와 시간이 필요합니다. 일주일에 한 번 정도 발생합니다. 반면 간단한 실수는 거의 매일 발생합니다.

ETL 프로세스 모니터링은 오류 위치가 가능한 한 정확하게 로그에 표시되고 문제의 원인을 찾는 데 최소한의 시간이 필요할 때 효과적입니다.

효과적인 모니터링

ETL 모니터링 프로세스에서 보고 싶었던 것은 무엇입니까?

시작 - 그가 일을 시작한 때,
소스 - 데이터 소스,
레이어 - 어떤 수준의 스토리지가 로드되고 있는지,
ETL 작업 이름 - 많은 작은 단계로 구성된 업로드 절차,
단계 번호 - 수행 중인 단계의 번호,
영향을 받는 행 - 이미 처리된 데이터의 양,
기간 초 - 걸리는 시간,
상태 - 모든 것이 정상인지 여부: OK, ERROR, RUNNING, HANGS
메시지 - 마지막으로 성공한 메시지 또는 오류 설명입니다.

항목의 상태에 따라 이메일을 보낼 수 있습니다. 다른 회원들에게 보내는 편지. 오류가 없으면 편지가 필요하지 않습니다.

따라서 오류가 발생한 경우 사고 위치가 명확하게 표시됩니다.

때때로 모니터링 도구 자체가 작동하지 않는 경우가 있습니다. 이 경우 보고서 작성을 기반으로 데이터베이스에서 직접 보기(보기)를 호출할 수 있습니다.

ETL 모니터링 테이블

ETL 프로세스 모니터링을 구현하려면 하나의 테이블과 하나의 보기로 충분합니다.

이렇게 하려면 다음으로 돌아갈 수 있습니다. 당신의 작은 창고 그리고 sqlite 데이터베이스에서 프로토타입을 생성합니다.

DDL 테이블

CREATE TABLE UTL_JOB_STATUS (
/* Table for logging of job execution log. Important that the job has the steps ETL_START and ETL_END or ETL_ERROR */
  UTL_JOB_STATUS_ID INTEGER NOT NULL PRIMARY KEY AUTOINCREMENT,
  SID               INTEGER NOT NULL DEFAULT -1, /* Session Identificator. Unique for every Run of job */
  LOG_DT            INTEGER NOT NULL DEFAULT 0,  /* Date time */
  LOG_D             INTEGER NOT NULL DEFAULT 0,  /* Date */
  JOB_NAME          TEXT NOT NULL DEFAULT 'N/A', /* Job name like JOB_STG2DM_GEO */
  STEP_NAME         TEXT NOT NULL DEFAULT 'N/A', /* ETL_START, ... , ETL_END/ETL_ERROR */
  STEP_DESCR        TEXT,                        /* Description of task or error message */
  UNIQUE (SID, JOB_NAME, STEP_NAME)
);
INSERT INTO UTL_JOB_STATUS (UTL_JOB_STATUS_ID) VALUES (-1);

DDL 보기/보고

CREATE VIEW IF NOT EXISTS UTL_JOB_STATUS_V
AS /* Content: Package Execution Log for last 3 Months. */
WITH SRC AS (
  SELECT LOG_D,
    LOG_DT,
    UTL_JOB_STATUS_ID,
    SID,
	CASE WHEN INSTR(JOB_NAME, 'FTP') THEN 'TRANSFER' /* file transfer */
	     WHEN INSTR(JOB_NAME, 'STG') THEN 'STAGE' /* stage */
	     WHEN INSTR(JOB_NAME, 'CLS') THEN 'CLEANSING' /* cleansing */
	     WHEN INSTR(JOB_NAME, 'DIM') THEN 'DIMENSION' /* dimension */
	     WHEN INSTR(JOB_NAME, 'FCT') THEN 'FACT' /* fact */
		 WHEN INSTR(JOB_NAME, 'ETL') THEN 'STAGE-MART' /* data mart */
	     WHEN INSTR(JOB_NAME, 'RPT') THEN 'REPORT' /* report */
	     ELSE 'N/A' END AS LAYER,
	CASE WHEN INSTR(JOB_NAME, 'ACCESS') THEN 'ACCESS LOG' /* source */
	     WHEN INSTR(JOB_NAME, 'MASTER') THEN 'MASTER DATA' /* source */
	     WHEN INSTR(JOB_NAME, 'AD-HOC') THEN 'AD-HOC' /* source */
	     ELSE 'N/A' END AS SOURCE,
    JOB_NAME,
    STEP_NAME,
    CASE WHEN STEP_NAME='ETL_START' THEN 1 ELSE 0 END AS START_FLAG,
    CASE WHEN STEP_NAME='ETL_END' THEN 1 ELSE 0 END AS END_FLAG,
    CASE WHEN STEP_NAME='ETL_ERROR' THEN 1 ELSE 0 END AS ERROR_FLAG,
    STEP_NAME || ' : ' || STEP_DESCR AS STEP_LOG,
	SUBSTR( SUBSTR(STEP_DESCR, INSTR(STEP_DESCR, '***')+4), 1, INSTR(SUBSTR(STEP_DESCR, INSTR(STEP_DESCR, '***')+4), '***')-2 ) AS AFFECTED_ROWS
  FROM UTL_JOB_STATUS
  WHERE datetime(LOG_D, 'unixepoch') >= date('now', 'start of month', '-3 month')
)
SELECT JB.SID,
  JB.MIN_LOG_DT AS START_DT,
  strftime('%d.%m.%Y %H:%M', datetime(JB.MIN_LOG_DT, 'unixepoch')) AS LOG_DT,
  JB.SOURCE,
  JB.LAYER,
  JB.JOB_NAME,
  CASE
  WHEN JB.ERROR_FLAG = 1 THEN 'ERROR'
  WHEN JB.ERROR_FLAG = 0 AND JB.END_FLAG = 0 AND strftime('%s','now') - JB.MIN_LOG_DT > 0.5*60*60 THEN 'HANGS' /* half an hour */
  WHEN JB.ERROR_FLAG = 0 AND JB.END_FLAG = 0 THEN 'RUNNING'
  ELSE 'OK'
  END AS STATUS,
  ERR.STEP_LOG     AS STEP_LOG,
  JB.CNT           AS STEP_CNT,
  JB.AFFECTED_ROWS AS AFFECTED_ROWS,
  strftime('%d.%m.%Y %H:%M', datetime(JB.MIN_LOG_DT, 'unixepoch')) AS JOB_START_DT,
  strftime('%d.%m.%Y %H:%M', datetime(JB.MAX_LOG_DT, 'unixepoch')) AS JOB_END_DT,
  JB.MAX_LOG_DT - JB.MIN_LOG_DT AS JOB_DURATION_SEC
FROM
  ( SELECT SID, SOURCE, LAYER, JOB_NAME,
           MAX(UTL_JOB_STATUS_ID) AS UTL_JOB_STATUS_ID,
           MAX(START_FLAG)       AS START_FLAG,
           MAX(END_FLAG)         AS END_FLAG,
           MAX(ERROR_FLAG)       AS ERROR_FLAG,
           MIN(LOG_DT)           AS MIN_LOG_DT,
           MAX(LOG_DT)           AS MAX_LOG_DT,
           SUM(1)                AS CNT,
           SUM(IFNULL(AFFECTED_ROWS, 0)) AS AFFECTED_ROWS
    FROM SRC
    GROUP BY SID, SOURCE, LAYER, JOB_NAME
  ) JB,
  ( SELECT UTL_JOB_STATUS_ID, SID, JOB_NAME, STEP_LOG
    FROM SRC
    WHERE 1 = 1
  ) ERR
WHERE 1 = 1
  AND JB.SID = ERR.SID
  AND JB.JOB_NAME = ERR.JOB_NAME
  AND JB.UTL_JOB_STATUS_ID = ERR.UTL_JOB_STATUS_ID
ORDER BY JB.MIN_LOG_DT DESC, JB.SID DESC, JB.SOURCE;

SQL 새 세션 번호를 얻을 수 있는지 확인

SELECT SUM (
  CASE WHEN start_job.JOB_NAME IS NOT NULL AND end_job.JOB_NAME IS NULL /* existed job finished */
	    AND NOT ( 'y' = 'n' ) /* force restart PARAMETER */
       THEN 1 ELSE 0
  END ) AS IS_RUNNING
  FROM
    ( SELECT 1 AS dummy FROM UTL_JOB_STATUS WHERE sid = -1) d_job
  LEFT OUTER JOIN
    ( SELECT JOB_NAME, SID, 1 AS dummy
      FROM UTL_JOB_STATUS
      WHERE JOB_NAME = 'RPT_ACCESS_LOG' /* job name PARAMETER */
	    AND STEP_NAME = 'ETL_START'
      GROUP BY JOB_NAME, SID
    ) start_job /* starts */
  ON d_job.dummy = start_job.dummy
  LEFT OUTER JOIN
    ( SELECT JOB_NAME, SID
      FROM UTL_JOB_STATUS
      WHERE JOB_NAME = 'RPT_ACCESS_LOG'  /* job name PARAMETER */
	    AND STEP_NAME in ('ETL_END', 'ETL_ERROR') /* stop status */
      GROUP BY JOB_NAME, SID
    ) end_job /* ends */
  ON start_job.JOB_NAME = end_job.JOB_NAME
     AND start_job.SID = end_job.SID

테이블 기능:

• 데이터 처리 절차의 시작과 끝 뒤에는 ETL_START 및 ETL_END 단계가 따라야 합니다.
• 오류의 경우 설명과 함께 ETL_ERROR 단계를 생성해야 합니다.
• 처리된 데이터의 양은 예를 들어 별표로 강조 표시되어야 합니다.
• force_restart=y 매개변수를 사용하여 동일한 프로시저를 동시에 시작할 수 있으며, 이것이 없으면 완료된 프로시저에만 세션 번호가 발행됩니다.
• 일반 모드에서는 동일한 데이터 처리 절차를 병렬로 실행할 수 없습니다.

테이블 작업에 필요한 작업은 다음과 같습니다.

• 실행 중인 ETL 프로시저의 세션 번호 가져오기
• 테이블에 로그 항목 삽입
• ETL 프로시저의 마지막 성공적인 레코드 가져오기

Oracle 또는 Postgres와 같은 데이터베이스에서 이러한 작업은 내장 함수로 구현될 수 있습니다. sqlite는 외부 메커니즘이 필요하며 이 경우 PHP로 프로토타입 제작.

출력

따라서 데이터 처리 도구의 오류 메시지는 매우 중요한 역할을 합니다. 그러나 문제의 원인을 빨리 찾기 위해 최적이라고 부르기는 어렵습니다. 절차 수가 XNUMX개에 가까워지면 프로세스 모니터링이 복잡한 프로젝트로 바뀝니다.

이 기사는 프로토타입의 형태로 문제에 대한 가능한 솔루션의 예를 제공합니다. 전체 작은 저장소 프로토타입은 gitlab에서 사용할 수 있습니다. SQLite PHP ETL 유틸리티.

출처 : habr.com