[CDC 프로젝트] Oracle DB to MySQL(1)

이전 포스팅에서 다뤘던 CDC에 있어 규모를 조금 줄여 직접 CDC를 진행해보려고 합니다.

 

CDC 기본적인 개념에 대해서 궁금하신 분들은 아래 포스팅을 참고하시기 바랍니다.

https://hyunily.tistory.com/41

CDC(Change Data Capture)란?

---

요구사항

이번 프로젝트에서 요구사항은 다음과 같습니다.

1. Oracle DB로부터 변경된 데이터를 읽는다.

2. 읽어온 데이터에서 적절한 정보만 가공한다.

2. 가공한 데이터를 MySQL에 동기화를 진행한다.

 

아키텍처

아키텍처는 다음과 같습니다.

 

 

1. Oracle로 부터 로그 데이터(DML)를 Spring Batch를 활용해서 읽어옵니다.

2. 읽어온 데이터를 Spring Batch에서 필요한 부분만 정제하여 Kafka Producer로 전송합니다.

3. Spring Batch에서 처리된 데이터를 Kafka로 게시(Publish)하고, Spring 기반 소비자(Consumer) 애플리케이션으로 데이터를 전달합니다.

4. Kafka에서 데이터를 구독한 후, 최종적으로 전달해야 할 데이터를 다시 Kafka에 전송합니다.

5. 변경 데이터를 동기화 하는 Spring 애플리케이션에서 전달받은 데이터를 MySQL에 반영합니다.

 

Oracle DB DML 로그 읽기

로그파일 버전 확인하기

로그를 읽기 앞서, 보통 DML 로그는 REDO 로그파일에 축적됩니다.

 

REDO 로그파일은 보통 C:/APP/USER/PRODUCT/21C/ORADATA/XE/ 하단 경로에 위치하고 있습니다.

경로는 Oracle DB 버전에 따라 이름이 상이할 수 있으니 확인해 보시길 바랍니다.

 

Spring Batch에서 로그 데이터를 읽어오기 전에, 일단 현재 DML내용을 축적하고 있는 로그파일이 어떤 건지 확인해야 합니다.

해당 로그파일을 확인하는 방법은 아래와 같습니다.

SELECT GROUP#, STATUS, ARCHIVED FROM V$LOG;

 

위 쿼리를 실행해보면 다음과 같은 결과를 얻을 수 있습니다.

STATUS가 CURRENT인 REDO 로그파일에 DML이 축적된다는 걸 확인할 수 있습니다.

GROUP에 표시되는 숫자가 REDO01.LOG, REDO02.LOG, REDO03.LOG 로그파일 명에 명시되어 있는 숫자입니다.

 

로그마이너 프로시저 실행

본격적인 로그파일 확인에 앞서, Oracle에서는 로그 마이너(LogMiner)를 사용하여 REDO 로그 파일을 분석합니다.

 

* LogMiner란?

- LogMiner는 Oracle의 REDO 로그를 분석하기 위한 유틸리티 도구입니다.

- REDO로그에는 데이터베이스의 모든 변경 사항(INSERT, UPDATE, DELETE 등)이 기록됩니다.

- LogMiner는 이를 분석하여 어떤 작업이 수행되었는지 확인할 수 있게 해 줍니다.

 

execute DBMS_LOGMNR.ADD_LOGFILE (LOGFILENAME => 'C:\APP\USER\PRODUCT\21C\ORADATA\XE\REDO03.LOG',OPTIONS => DBMS_LOGMNR.NEW);

execute DBMS_LOGMNR.START_LOGMNR (OPTIONS => DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG);

 

쿼리 내용은 다음과 같습니다.

 

DBMS_LOGMNR.ADD_LOGFILE : LogMiner에 분석할 로그 파일을 추가하는 작업

 

OPTIONS => DBMS_LOGMNR.NEW : 기존에 등록된 로그 파일을 모두 삭제하고, 새로운 로그파일을 추가

 

DBMS_LOGMNR.START_LOGMNR : LogMiner의 세션을 시작

 

OPTIONS => DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG : 데이터를 어떻게 해석할지 정하는 부분으로, Oracle 데이터를 딕셔너리를 사용하여 데이터베이스 메타데이터를 가져옴

 

로그 데이터 읽기

SELECT OPERATION, SEG_OWNER, TABLE_NAME, SQL_REDO FROM V$LOGMNR_CONTENTS WHERE TABLE_NAME = 'INTERACTION';

 

간단하게 로그 데이터를 위 쿼리를 활용해서 읽을 수 있습니다.

수많은 로그 데이터 중, SELECT 한 데이터는 아래와 같습니다.

 

OPERATION : 트랜잭션의 작업 유형을 나타냅니다. INSERT, UPDATE, DELETE, DDL 등 어떤 작업이 실행되었는지를 담고 있습니다.

SEG_OWNER : 테이블의 소유자(Schema)를 담고 있습니다.

TABLE_NAME : 작업이 실행된 테이블의 이름을 담고 있습니다.

SQL_REDO : 로그에 기록된 작업을 재실행하기 위한 SQL문을 담고 있습니다.

 

이렇게 쿼리를 실행하고 나면 아래 사진과 같이 모든 변경사항을 확인할 수 있습니다.

 

이외에도 더 많은 정보를 가져올 수 있으니, 전체 조회로 어떤 데이터를 담고 있는지, 필요한 데이터는 어떤 게 있는지 확인하시고 활용하시기 바랍니다.

 

Spring Batch에서 Oracle 로그 데이터 읽어오기

위 과정에서는 Oracle DB에서 직접 설정과 조회를 진행했지만, 해당 과정을 결국 Spring Batch에서 읽어와야 합니다.

 

Spring Batch에서 일단 읽어오는 과정은 다음과 같을 겁니다.

1. 현재 활성화된 REDO 로그파일 경로 조회

2. DBMS_LOGMNR.ADD_LOGFILE 실행

3. DBMS_LOGMNR.START_LOGMNR 실행

4. V$LOGMNR_CONTENTS 테이블에서 데이터 가져오기

5. 결과 확인

 

그럼 단계별로 살펴보겠습니다.

1. 현재 활성화된 REDO 로그파일 경로 조회

기존 Oracle에서는 전체 로그파일을 조회해서 CURRENT 상태인 로그파일을 확인했지만, 이젠 처음부터 조회를 CURRENT 상태인 로그파일의 경로를 가져올 겁니다.

public String getCurrentRedoLogFile(JdbcTemplate jdbcTemplate) {
    String query = """
                SELECT B.MEMBER AS LOGFILE_PATH
                FROM V$LOG A
                JOIN V$LOGFILE B ON A.GROUP# = B.GROUP#
                WHERE A.STATUS = 'CURRENT'
            """;

    return jdbcTemplate.queryForObject(query, String.class);
}

 

2. DBMS_LOGMNR.ADD_LOGFILE 실행

현재 활성화된 로그파일의 경로를 활용해서 로그마이너 ADD_LOGFILE을 실행합니다.

@Bean
public Tasklet readOracleLogStepTasklet() {
    return ((contribution, chunkContext) -> {

        String currentRedoLogFile = getCurrentRedoLogFile(jdbcTemplate);

        String addLogFileSql = String.format(
                "BEGIN DBMS_LOGMNR.ADD_LOGFILE(" +
                        "LOGFILENAME => '%s', " +
                        "OPTIONS => DBMS_LOGMNR.NEW); END;", currentRedoLogFile
        );
        jdbcTemplate.execute(addLogFileSql);
        
                                         ...
    });
}

 

3. DBMS_LOGMNR.START_LOGMNR 실행

적절한 옵션을 설정해 로그마이너를 실행합니다.

@Bean
public Tasklet readOracleLogStepTasklet() {
    return ((contribution, chunkContext) -> {

                                         ...
        
        String startLogMinerSql = "BEGIN DBMS_LOGMNR.START_LOGMNR(" +
                    "OPTIONS => DBMS_LOGMNR.DICT_FROM_ONLINE_CATALOG); END;";
        jdbcTemplate.execute(startLogMinerSql);
        
                                         ...
    });
}

 

4. V$LOGMNR_CONTENTS 테이블에서 데이터 가져오기

필요한 정보를 고르고, 어떤 테이블로부터 추적할지 설정한 후 데이터를 가져옵니다.

@Bean
public Tasklet readOracleLogStepTasklet() {
    return ((contribution, chunkContext) -> {

                                         ...
        
        String selectLogContentsSql = "SELECT OPERATION, SEG_OWNER, TABLE_NAME, SQL_REDO " +
                                        "FROM V$LOGMNR_CONTENTS " +
                                        "WHERE TABLE_NAME = 'INTERACTION'";
        List<Map<String, Object>> logContentsResults = jdbcTemplate.queryForList(selectLogContentsSql);
        
                                         ...
    });
}

 

5. 결과 확인

조회한 데이터를 담았던 logContentsResults를 확인해 보면, Oracle에서 쿼리로 확인했던 내용들이 담겨있는 걸 확인할 수 있습니다.

 

---

 

Oracle에서 로그 데이터를 조회하는 방법부터 실제 Spring Batch에서 불러오는 방법까지의 과정을 다뤄봤습니다.

 

아키텍처과정에서 Spring Batch에서 데이터를 불러오는 과정까지의 내용입니다.

 

이 이후 과정에서는 데이터 적재 및 정제 과정을 담은 포스팅을 다뤄보도록 하겠습니다.