본문으로 건너뛰기
최종 수정: 에

SQL 매핑을 사용한 MySQL CDC

실험적 기능

Doris 지속 적재는 실험적 기능이며 Doris 4.1.5부터 지원됩니다.

SQL 매핑 동기화는 Job + CDC Stream TVF를 통해 구현되며, 이미 존재하는 Doris 테이블을 대상으로 합니다(INSERT INTO tbl SELECT * FROM cdc_stream(...)). Doris SQL의 표현력을 활용하면 동기화 파이프라인에서 열 매핑, 필터링 및 데이터 변환을 수행할 수 있습니다. 데이터 처리가 필요한 실시간 동기화 시나리오에 적합합니다.

Flink CDC의 읽기 기능을 통합하여 Doris는 MySQL에서 변경 로그(Binlog)를 읽고 소스 테이블에서 대상 테이블로 전체 + 증분 동기화를 수행합니다. Doris가 다운스트림 테이블을 자동으로 만들고 데이터베이스별로 테이블 그룹을 동기화하도록 하려면 MySQL CDC 자동 테이블 생성 동기화를 참조하세요.

적용 시나리오​

  • MySQL 단일 테이블을 Doris로 지속 동기화하며 대상 테이블 스키마가 이미 계획된 경우
  • 동기화 중 열 제거, 열 매핑, 필드 이름 변경 또는 데이터 변환이 필요한 경우
  • 엔드투엔드 exactly-once 의미 체계가 필요한 실시간 데이터 통합

사전 요구 사항​

항목설명
Doris 버전4.1.5 이상
테이블 유형업스트림 소스 테이블에는 기본 키가 있어야 하며 Doris 대상 테이블은 Unique Key 모델을 사용해야 합니다
사용자 권한Load 권한이 필요합니다
MySQL 구성MySQL 측에서 Binlog를 활성화해야 합니다. 구성 가이드를 참조하세요
의미 체계 보장exactly-once 의미 체계를 지원합니다

빠른 시작​

다음 최소 실행 가능 예제에서는 전체 흐름인 작업 생성과 상태 확인을 보여 줍니다.

1단계: 가져오기 작업 생성​

CREATE STREAMING JOB을 사용하여 지속 가져오기 작업을 생성합니다.

CREATE JOB mysql_single_sync
ON STREAMING
DO
INSERT INTO db1.tbl1
SELECT * FROM cdc_stream(
    "type" = "mysql",
    "jdbc_url" = "jdbc:mysql://127.0.0.1:3306",
    "driver_url" = "mysql-connector-java-8.0.25.jar",
    "driver_class" = "com.mysql.cj.jdbc.Driver",
    "user" = "root",
    "password" = "123456",
    "database" = "source_db",
    "table" = "source_table",
    "offset" = "initial"
)

2단계: 가져오기 상태 확인​

select * from jobs("type"="insert") where ExecuteType = "STREAMING";

3단계: 작업 운영​

일반적인 작업(일시 중지, 재개, 삭제, Task 보기 등)은 지속 가져오기 개요를 참조하세요.

매개변수 참조​

데이터 소스 매개변수​

CDC Stream TVF는 다음 MySQL 데이터 소스 매개변수를 지원합니다.

매개변수필수기본값설명
type예-데이터 소스 유형입니다. mysql로 설정합니다.
jdbc_url예-MySQL JDBC 연결 문자열입니다.
driver_url예-JDBC 드라이버 jar 경로입니다. 파일 이름, 로컬 절대 경로 또는 HTTP URL을 지원합니다. 자세한 내용은 JDBC Catalog 개요를 참조하세요.
driver_class예-JDBC 드라이버 클래스 이름입니다(예: com.mysql.cj.jdbc.Driver).
user예-데이터베이스 사용자 이름입니다.
password예-데이터베이스 비밀번호입니다.
database예-MySQL 데이터베이스 이름입니다.
table예-동기화할 테이블 이름입니다. 각 SQL Mapping 작업은 하나의 소스 테이블만 지원합니다.
offset예-시작 오프셋입니다. initial: 전체 및 증분 동기화, snapshot: 전체 동기화만 수행, earliest: 사용 가능한 가장 이른 Binlog 오프셋부터 시작, latest: 작업 시작 후 변경 사항만 동기화합니다. 정확한 JSON 오프셋을 지정할 수도 있습니다(예: {"file":"binlog.000001","pos":"154"} 또는 {"gtids":"<gtid_set>"}).
snapshot_split_size아니요40960행 단위 분할 크기입니다. 전체 동기화 중 테이블이 여러 split으로 나뉩니다. 양의 정수여야 합니다.
snapshot_parallelism아니요1전체 동기화 단계의 병렬 처리 수, 즉 한 번에 Task가 예약하는 최대 split 수입니다. 양의 정수여야 합니다.
skip_snapshot_backfill아니요false스냅샷 중 Binlog 백필을 건너뛸지 여부입니다. true로 설정하면 at-least-once 의미 체계를 사용합니다.
server_id아니요자동 생성MySQL CDC reader의 서버 ID입니다. 5400과 같은 단일 값이나 5400-5408과 같은 폐구간을 지원합니다. 구간 너비는 snapshot_parallelism 이상이어야 합니다.
ssl_mode아니요disableSSL 모드입니다. 유효한 값은 disable, require, verify-ca입니다.
ssl_rootcert조건부 필수-FILE:<file_name> 형식의 CA 인증서 파일입니다. ssl_mode가 verify-ca인 경우 필요합니다. 먼저 CREATE FILE을 사용하여 파일을 업로드하세요.
include_delete_sign아니요falseTVF가 __DORIS_DELETE_SIGN__ 열을 추가로 출력할지 여부입니다. 업스트림 DELETE 작업을 Doris 기본 키 테이블의 삭제로 동기화하려면 true로 설정합니다.

include_delete_sign을 활성화하면 대상 테이블은 Merge-on-Write Unique Key 테이블이어야 하며 INSERT 대상 열 목록과 SELECT 목록 모두에서 __DORIS_DELETE_SIGN__을 명시적으로 매핑해야 합니다.

CREATE JOB mysql_cdc_with_delete
ON STREAMING
DO
INSERT INTO db1.target_table (id, value, __DORIS_DELETE_SIGN__)
SELECT id, value, __DORIS_DELETE_SIGN__
FROM cdc_stream(
    "type" = "mysql",
    "jdbc_url" = "jdbc:mysql://127.0.0.1:3306",
    "driver_url" = "mysql-connector-java-8.0.25.jar",
    "driver_class" = "com.mysql.cj.jdbc.Driver",
    "user" = "root",
    "password" = "123456",
    "database" = "source_db",
    "table" = "source_table",
    "offset" = "initial",
    "include_delete_sign" = "true"
);

Job 구성 매개변수​

다음 매개변수는 CREATE JOB ... PROPERTIES (...)를 통해 설정합니다.

매개변수기본값설명
max_interval10업스트림에 새 데이터가 없을 때의 유휴 예약 간격(초)입니다. 1 이상인 정수여야 합니다.
compute_group현재 세션 또는 사용자 기본 compute groupcompute-storage 분리 모드에서만 지원됩니다. 작업이 실행되는 compute group을 지정합니다. 사용자에게 해당 compute group에 대한 USAGE 권한이 있어야 합니다.
session.<variable_name>해당 세션 변수의 기본값INSERT 작업의 세션 변수를 설정합니다. 적재 변수는 Insert Into Select를 참조하세요.

CDC 위치를 재설정하는 데 사용되는 Job Property offset과 ALTER JOB을 통한 전체 제한 사항은 공통 작업 적재 구성 매개변수를 참조하세요. 새 작업의 초기 위치는 cdc_stream(...)에서 설정해야 합니다.

FAQ​

Q1: SQL 매핑 동기화와 자동 테이블 생성 동기화의 차이점은 무엇인가요?

  • SQL 매핑 동기화: 대상 Doris 테이블을 미리 생성해야 합니다. 열 매핑과 데이터 변환을 지원하며 세밀한 처리가 필요한 시나리오에 적합합니다.
  • 자동 테이블 생성 동기화: Doris가 다운스트림 테이블을 자동으로 생성하고 데이터베이스 단위로 전체 동기화합니다. 자세한 내용은 MySQL CDC 자동 테이블 생성 동기화를 참조하세요.

Q2: 기본 키가 아닌 테이블을 대상 테이블로 지원하나요?

업스트림 소스 테이블에는 기본 키가 있어야 하며 Doris 대상 테이블은 현재 Unique Key 모델을 사용해야 합니다.

Q3: 기록된 전체 데이터 없이 증분 데이터만 동기화하려면 어떻게 해야 하나요?

offset 매개변수를 latest로 설정합니다. 작업은 전체 동기화 단계를 건너뛰고 Binlog 증분 데이터만 동기화합니다.

Q4: 전체 동기화가 너무 느릴 때 어떻게 최적화하나요?

전체 동기화 단계의 처리량을 높이려면 다음 두 매개변수를 조정할 수 있습니다.

  • snapshot_split_size: split당 행 수를 늘립니다.
  • snapshot_parallelism: Task당 예약되는 split의 병렬 처리 수를 늘립니다.

snapshot_parallelism을 조정할 때 server_id 범위도 구성되어 있다면 범위 내 ID 수가 병렬 처리 수보다 적지 않은지 확인하세요.

Report issue