Lakehouse Federation 병렬 읽기 — JDBC 커넥터로 큰 테이블 빠르게 가져오기

Lakehouse Federation은 PostgreSQL·MySQL·SQL Server 같은 외부 데이터베이스를 Unity Catalog의 foreign catalog로 붙여, 데이터를 복사하지 않고 Databricks에서 바로 쿼리하게 해 줍니다. 작은 테이블을 조회하거나 필터가 잘 푸시다운되는 쿼리라면 별다른 설정 없이도 잘 동작합니다.

문제는 큰 테이블을 통째로, 또는 넓은 범위로 읽을 때입니다. 수천만 행짜리 주문 테이블을 Delta로 옮기려고 CREATE TABLE ... AS SELECT를 실행하면, 클러스터에 익스큐터가 수십 개 있어도 작업 하나만 길게 돌아가는 모습을 보게 됩니다. JDBC 커넥터가 기본적으로 원본 DB에 연결 하나를 열고 쿼리 하나로 전체를 읽기 때문입니다.

Databricks 문서 Lakehouse Federation performance recommendations의 Enable parallel reads for JDBC connectors 절이 이 문제를 다룹니다. 이 글은 그 내용을 정리하고, 실제로 적용할 때 판단해야 하는 지점(경계값, 파티션 컬럼, 파티션 수, 원본 DB 부하)을 보탭니다.

1. 기본 동작: 연결 하나, 쿼리 하나

JDBC 전송 프로토콜을 쓰는 커넥터는 별도 설정이 없으면 원격 쿼리 하나를 하나의 작업(task)이 처리합니다. 결과가 아무리 커도 한 연결로 순서대로 받아 오므로, 클러스터 크기를 키워도 읽기 속도는 거의 변하지 않습니다.

여기서 자주 헷갈리는 옵션이 fetchSize입니다.

SELECT * FROM mySqlCatalog.schema.table WITH ('fetchSize' 100000)

fetchSize는 왕복 한 번에 가져올 최대 행 수를 정합니다. 대부분의 JDBC 커넥터는 기본적으로 결과를 한꺼번에 가져오므로 메모리 부족이 날 수 있는데, fetchSize를 지정하면 배치 단위로 나눠 읽어 이를 막습니다. 문서는 배치가 너무 작으면 전체 실행 시간이 길어지므로 100,000처럼 큰 값을 권장합니다.

다만 문서가 명시하듯 fetchSize는 배치로 나눠 읽게 할 뿐 병렬로 읽게 하지는 않습니다. 메모리 문제를 푸는 옵션이지 속도를 높이는 옵션이 아닙니다. 속도를 높이려면 쿼리 자체를 여러 조각으로 나눠야 하고, 그것이 병렬 읽기입니다.

2. 병렬 읽기 켜기: 옵션 4개

병렬 읽기는 쿼리를 숫자 컬럼의 범위로 나눠, 여러 익스큐터가 각자 한 범위씩 동시에 읽게 하는 방식입니다. 네 가지 옵션을 함께 지정합니다.

문서의 예시는 id가 1~1,000이고 id마다 한 행씩 있다고 가정해, 쿼리를 4개 파티션으로 나눕니다. 파티션 하나가 id 약 250개 범위를 맡습니다.

SELECT * FROM mySqlCatalog.schema.table WITH (
  'numPartitions' 4,
  'partitionColumn' 'id',
  'lowerBound' 1,
  'upperBound' 1000
)

옵션은 테이블 참조 뒤의 WITH (...) 절에 붙이므로, 조회뿐 아니라 Delta로 옮기는 CTAS에도 그대로 쓸 수 있습니다.

CREATE TABLE main.bronze.orders AS
SELECT * FROM pg_catalog.public.orders WITH (
  'numPartitions' 8,
  'partitionColumn' 'order_id',
  'lowerBound' 1,
  'upperBound' 48000000
);

지원 커넥터와 컴퓨트 요구 사항

• 지원 커넥터: Databricks, Microsoft SQL Server, Microsoft Azure Synapse, MySQL, Oracle, PostgreSQL, Redshift, Salesforce Data 360, Teradata
• 컴퓨트: Pro 또는 Serverless SQL 웨어하우스, 혹은 Databricks Runtime 16.4 LTS 이상

Snowflake는 이 목록에 없습니다. Snowflake 커넥터는 원래 여러 파티션으로 나눠 가져오며, 파티션 크기를 partition_size_in_mb 옵션(기본 100 MB, 압축 해제 기준 권장 크기)으로 조정합니다.

3. 경계값은 행을 거르지 않는다

병렬 읽기에서 가장 많이 오해하는 부분입니다. 문서는 이렇게 못 박습니다.

lowerBound와 upperBound는 파티션 간격을 정하는 데만 쓰이며, 테이블의 행을 거르지 않습니다. 테이블의 모든 행이 파티션으로 나뉘어 반환됩니다.

즉 lowerBound 1, upperBound 1000이라고 써도 id가 5,000인 행은 빠지지 않습니다. 경계 밖의 행은 첫 파티션과 마지막 파티션이 떠안습니다. Apache Spark JDBC 데이터 소스의 분할 방식으로 보면, numPartitions 4, lowerBound 0, upperBound 1000일 때 간격은 250이고 각 파티션에는 대략 다음 조건이 붙습니다.

첫 파티션은 아래쪽이, 마지막 파티션은 위쪽이 열려 있으므로 결과가 빠지는 일은 없습니다. 대신 경계를 실제 데이터 범위와 다르게 잡으면 일이 한 파티션에 몰립니다.

위 그림의 B처럼 실제 id가 10,000까지 있는데 upperBound를 1,000으로 두면, 앞의 세 파티션은 금방 끝나고 마지막 파티션 하나가 전체의 92.5%를 읽습니다. 쿼리는 정상적으로 끝나고 결과도 맞기 때문에, 느려진 원인이 설정이라는 사실을 알아채기 어렵습니다.

반대로 numPartitions를 크게 잡고 경계를 넉넉하게(예: upperBound를 실제 최댓값의 10배로) 두면, 대부분의 파티션이 빈 범위를 읽느라 원본 DB에 쓸모없는 쿼리만 날리게 됩니다.

파티션 조건의 정확한 형태는 Databricks 문서에 나와 있지 않습니다. 위 표는 같은 이름의 옵션을 쓰는 Spark JDBC 데이터 소스의 동작을 기준으로 한 설명이므로, 실제 조건은 6절의 방법으로 원본 DB에서 확인하세요.

4. 파티션 컬럼 고르기

문서가 꼽는 조건은 세 가지입니다.

1. 숫자형 컬럼
2. 범위 안에서 고르게 분포
3. 성능을 위해 인덱스가 있는 컬럼

세 번째가 생각보다 중요합니다. 파티션마다 WHERE id >= ... AND id < ... 같은 범위 조건이 원본 DB로 가는데, 인덱스가 없으면 파티션 수만큼 전체 테이블 스캔이 일어납니다. 8개로 나누면 원본 DB는 같은 테이블을 8번 통째로 읽습니다. 병렬 읽기로 Databricks 쪽은 빨라져도 원본 DB가 버티지 못합니다.

실무에서 자주 만나는 경우를 정리하면 다음과 같습니다.

날짜 컬럼을 쓰고 싶다면

날짜를 기준으로 나누고 싶다면 원본 DB에 뷰를 만들어 숫자 컬럼으로 노출합니다. 이때 20261030 같은 YYYYMMDD 정수는 피하세요. 10월 31일(20261031) 다음이 11월 1일(20261101)이라 값 사이에 빈 구간이 생겨 분포가 고르지 않습니다. 기준일로부터 지난 일수처럼 연속된 값을 쓰는 편이 낫습니다. PostgreSQL이라면 다음과 같습니다.

-- 원본 PostgreSQL에서
CREATE VIEW reporting.events_by_day AS
SELECT e.*, (e.created_at::date - DATE '1970-01-01') AS created_day
FROM public.events e;

다만 이렇게 계산한 컬럼에는 인덱스가 바로 걸리지 않으므로, created_at에 인덱스가 있어도 범위 조건이 인덱스를 타는지 원본 DB의 실행 계획으로 확인해야 합니다. 안 탄다면 파티션마다 전체 스캔이 일어나는 앞의 문제로 돌아갑니다.

5. 경계값과 파티션 수 정하기

경계값: 실제 최솟값·최댓값을 먼저 조회

경계값은 추측하지 말고 원본에서 조회합니다.

-- 원본 DB에서 (PostgreSQL 예시)
SELECT MIN(order_id), MAX(order_id), COUNT(*) FROM public.orders;

최댓값이 하나만 튀는 경우(예: 테스트용으로 넣은 id = 999999999)라면, 그 값을 upperBound로 쓰는 순간 대부분의 파티션이 빈 범위를 맡습니다. 분포를 한 번 확인해 두면 안전합니다.

-- 원본 PostgreSQL에서: 1~48,000,000 범위를 8칸으로 나눴을 때 칸별 행 수
SELECT width_bucket(order_id, 1, 48000001, 8) AS bucket, COUNT(*)
FROM public.orders
GROUP BY 1
ORDER BY 1;

칸별 행 수가 비슷하면 그대로 쓰고, 특정 칸에 몰려 있다면 경계를 조정하거나 다른 컬럼을 찾습니다. 데이터가 계속 들어오는 테이블이라면 upperBound는 작업 시점의 최댓값보다 약간 크게 잡아도 됩니다. 넘치는 행은 마지막 파티션이 읽으므로 결과가 빠지지는 않습니다.

파티션 수: Databricks 쪽보다 원본 DB가 상한

numPartitions는 Databricks 입장에서는 동시에 돌릴 작업 수지만, 원본 DB 입장에서는 동시에 열리는 연결 수이자 동시에 실행되는 쿼리 수입니다. 그래서 상한은 대개 원본 DB가 정합니다.

• 원본 DB의 여유: 최대 연결 수(PostgreSQL max_connections 등), 다른 업무 트래픽, CPU·디스크 I/O 여유를 고려합니다. 운영 OLTP DB라면 4~8개처럼 작게 시작해 원본 DB 지표를 보며 늘리는 편이 안전합니다.
• 읽기 전용 리플리카: 가능하면 운영 기본 DB 대신 읽기 전용 리플리카를 바라보는 Unity Catalog 연결(connection)과 foreign catalog를 따로 만들어 쓰세요. 대량 추출이 운영 트래픽과 경쟁하지 않습니다.
• Databricks 쪽 코어 수: 파티션 수가 클러스터의 동시 작업 슬롯보다 많으면 몇 차례에 나눠 실행됩니다. 슬롯보다 훨씬 많이 잡아도 그만큼 빨라지지는 않습니다.
• 파티션당 크기: 파티션 하나가 너무 크면 메모리 부족이 날 수 있습니다. 문서는 fetchSize와 병렬 읽기를 따로 설명하지만, 둘 다 JDBC 읽기 옵션이므로 한 WITH 절에 함께 지정해 파티션마다 배치로 읽게 해 볼 수 있습니다. 조합 예시는 문서에 없으니 사용하는 환경에서 동작을 확인하세요.

SELECT * FROM pg_catalog.public.orders WITH (
  'numPartitions' 8,
  'partitionColumn' 'order_id',
  'lowerBound' 1,
  'upperBound' 48000000,
  'fetchSize' 100000
)

6. 병렬로 읽혔는지 확인하기

설정이 먹었는지는 양쪽에서 확인합니다.

• Databricks 쪽: 쿼리 프로필이나 Spark UI에서 스캔 단계의 작업 수가 numPartitions와 같은지, 작업별 소요 시간이 비슷한지 봅니다. 한 작업만 유난히 길다면 3절의 쏠림입니다.
• 원본 DB 쪽: 실행 중에 세션 목록을 보면 범위 조건이 다른 쿼리가 파티션 수만큼 동시에 돌고 있어야 합니다. PostgreSQL이라면 다음과 같습니다.

SELECT pid, state, now() - query_start AS running_for, left(query, 120)
FROM pg_stat_activity
WHERE datname = current_database() AND state = 'active';

• 원격 쿼리 자체: EXPLAIN FORMATTED로 원본 DB에 보낼 쿼리를 미리 볼 수 있습니다. 다만 문서가 밝히듯 적응형 쿼리 실행(adaptive query execution) 때문에 실제 쿼리와 다를 수 있으므로, 최종 확인은 원본 DB 쪽에서 하는 것이 확실합니다.

7. 뷰를 거치면 병렬 읽기가 꺼진다

문서가 따로 경고하는 제약입니다.

Databricks에서 만든 뷰가 federated 테이블을 참조하는 경우, 그 뷰를 쿼리할 때는 병렬 읽기가 지원되지 않습니다. 뷰는 원본 데이터베이스에 만드세요.

Unity Catalog에 편의용 뷰를 만들어 두고 그 뷰에 WITH 옵션을 붙이면 기대한 병렬 읽기가 일어나지 않습니다. 같은 뷰를 원본 DB에 만들고, foreign catalog를 통해 그 뷰를 직접 쿼리합니다.

-- 원본 데이터베이스에서
CREATE VIEW my_source_database_view AS SELECT * FROM source_database_schema.table;

-- Databricks에서
SELECT * FROM myfederated_catalog.schema.my_source_database_view WITH (
  'numPartitions' 4,
  'partitionColumn' 'id',
  'lowerBound' 1,
  'upperBound' 1000
)

4절의 날짜 컬럼 변환처럼 파티션 컬럼을 가공해야 할 때도 같은 방식, 즉 원본 DB 뷰를 쓰면 됩니다.

8. 함께 보면 좋은 최적화

같은 문서의 다른 권고도 병렬 읽기와 함께 효과를 냅니다. 읽을 행 자체를 줄이는 쪽이 병렬화보다 먼저입니다.

• 조건은 AND로 묶기: 푸시다운되지 않는 조건(예: MySQL에서 ILIKE)이 있어도, AND로 묶인 다른 조건(예: 날짜 비교)은 원본 DB로 푸시다운되어 가져오는 행을 줄입니다.
• 조인 푸시다운: 조인을 원본 DB에서 처리해 네트워크로 옮기는 데이터를 줄입니다. Redshift·Snowflake·BigQuery는 정식 지원(GA)이며 기본으로 켜져 있고, Oracle·PostgreSQL·MySQL·SQL Server·Teradata는 퍼블릭 프리뷰입니다(Databricks Runtime 17.2 이상, 프리뷰 설정 필요). inner·left outer·right outer 조인만 지원하며, 조인 아래에 limit·offset·집계가 있으면 푸시다운되지 않습니다.

9. 적용 체크리스트

1. 필요한 경우인가? 큰 테이블 전체 또는 넓은 범위를 읽는 작업(초기 적재, CTAS, 대량 추출)인지 확인합니다. 필터로 몇천 행만 가져오는 쿼리에는 필요 없습니다.
2. 컴퓨트 요건: Databricks Runtime 16.4 LTS 이상 또는 Pro·Serverless SQL 웨어하우스인지 확인합니다.
3. 파티션 컬럼: 숫자형이고, 고르게 분포하고, 인덱스가 있는 컬럼을 고릅니다. 보통 자동 증가 기본 키입니다.
4. 경계값: 원본에서 MIN/MAX를 조회하고, 튀는 값이 있으면 분포를 확인합니다.
5. 파티션 수: 원본 DB의 연결·부하 여유에 맞춰 작게 시작합니다. 운영 DB라면 읽기 전용 리플리카를 씁니다.
6. 뷰: Databricks 뷰가 아니라 원본 DB 뷰를 거치게 합니다.
7. 확인: 작업 수와 작업별 소요 시간, 원본 DB의 동시 쿼리를 봅니다.

병렬 읽기는 Databricks 쪽 설정 한 줄이지만, 실제 부하는 원본 데이터베이스가 받습니다. 파티션 수를 정할 때는 클러스터 크기보다 원본 DB가 동시 쿼리를 몇 개까지 버틸 수 있는지를 먼저 보세요.

참고 자료

• Databricks 문서: Lakehouse Federation performance recommendations — Enable parallel reads for JDBC connectors
• Apache Spark 문서: JDBC To Other Databases (partitionColumn·lowerBound·upperBound·numPartitions 옵션 설명)