Spark 동시 JDBC 데이터 읽기

Jan 27 2023

JDBC 소스에서 데이터를 읽기 위해 프로젝트에서 Spark를 구현하고 최적의 셔플 파티션 수, 드라이버 및 실행기 인스턴스에 대한 메모리 할당, 실행기 코어 수 및 그 모든 재미있는 것들을 결정하는 프로세스를 거친 적이 있습니까? ? 위 코드의 문제점은 무엇입니까? 글쎄요...너무 느립니다 :) 위의 예시처럼 스파크에서 SQL 쿼리를 실행하면 하나의 스레드만 사용하게 되고(코드를 실행하는 동안 하나의 태스크만 실행되는 것을 스파크 UI에서 볼 수 있음) 생성된 데이터 프레임은 파티션이 하나 있습니다. 우리 대부분은 병렬 처리를 위해 ETL/ELT 파이프라인에서 Spark를 사용하므로 하나의 스레드만 사용하여 JDBC 소스에서 데이터를 읽는 것은 아마도 우리가 원하는 것이 아닐 것입니다.

JDBC 소스에서 데이터를 읽기 위해 프로젝트에서 Spark를 구현하고 최적의 셔플 파티션 수, 드라이버 및 실행기 인스턴스에 대한 메모리 할당, 실행기 코어 수 및 그 모든 재미있는 것들을 결정하는 프로세스를 거친 적이 있습니까? ?

jdbcDF = spark.read \
    .format("jdbc") \
    .option("driver", "org.postgresql.Driver") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("user", os.environ['user']) \
    .option("password", os.environ['pass']) \
    .option("query", query) \
    .load()

위 코드의 문제점은 무엇입니까 ? 글쎄요...너무 느립니다 :) 위의 예시처럼 스파크에서 SQL 쿼리를 실행하면 하나의 스레드만 사용하게 되고(코드를 실행하는 동안 하나의 태스크만 실행되는 것을 스파크 UI에서 볼 수 있음) 생성된 데이터 프레임은 파티션이 하나 있습니다.

우리 대부분은 병렬 처리를 위해 ETL/ELT 파이프라인에서 Spark를 사용하므로 하나의 스레드만 사용하여 JDBC 소스에서 데이터를 읽는 것은 아마도 우리가 원하는 것이 아닐 것입니다. 데이터가 이제 1개의 파티션에 저장되기 때문에 쿼리를 실행한 후 데이터 프레임을 다시 분할해야 한다는 것은 말할 것도 없습니다. 그렇다면 실제로 Spark에서 어떻게 데이터를 동시에 읽을 수 있을까요?

# pass the original query to dbtable as a subquery 
final_query = f'({query}) as q'

# depends on your usecase, how many executor cores you have in the cluster
# what do you plan to do with the dataframe downstream, data size etc
partitionsCount = 100

# I'm hardcoding the date values here, but I'll show below how to identify
# these values at run time if you don't know them upfront
min_date = '2020-01-01'
max_date = '2022-12-31'

jdbcDF = spark.read \
    .format("jdbc") \
    .option("driver", "org.postgresql.Driver") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("user", os.environ['user']) \
    .option("password", os.environ['pass']) \
    .option("dbtable", final_query) \
    .option("numPartitions", partitionsCount) \
    .option("partitionColumn", "date") \
    .option("lowerBound", f"{min_date}") \
    .option("upperBound", f"{max_date}") \
    .load()

dbtable: Spark의 공식 문서에 따르면 데이터를 동시에 읽고 필요에 따라 분할하려는 경우 더 이상
쿼리 옵션을 사용할 수 없으며 대신 dbtable 옵션 을 통해 SQL 쿼리를 전달해야 합니다 . 이 두 옵션(이름 제외)의 유일한 차이점은 이제 쿼리를 괄호로 묶고 별칭을 가져야 한다는 점입니다. 기본적으로 원래 쿼리를 하위 쿼리로 전달합니다.
numPartitions:
테이블 읽기 및 쓰기에서 병렬 처리에 사용할 수 있는 최대 파티션 수입니다. 이것은 또한 동시 JDBC 연결의 최대 수를 결정합니다. 쓸 파티션 수가 이 한도를 초과하면 쓰기 전에 coalesce(numPartitions)를 호출하여 이 한도까지 줄입니다. ( 출처 ) EMR 클러스터에서 내 것을 100으로 설정했지만 파티션 번호를 코어 수의 1~4배 사이로 설정하는 것이 좋습니다. 하지만 고정된 숫자는 아니므로 항상 다른 값으로 테스트하여 어떤 것이 적합한지 확인하세요.
partitionColumn:
이것은 아마도 가장 중요한 매개변수이며 쿼리 실행 시간에 가장 큰 차이를 만드는 매개변수일 것입니다. 명심해야 할 몇 가지 사항:
- 문제의 열은 숫자, 날짜 또는 타임스탬프여야 합니다
. - 가능한 최상의 성능을 위해 열 값은 가능한 한 균등하게 분산되고 높은 카디널리티를 가져야 합니다. 우리는 데이터가 왜곡된 열을 피하려고 합니다. 이에 대해서는 아래에서 자세히 설명하겠습니다.
- 위의 속성을 가진 컬럼이 여러 개인 경우 인덱싱된 컬럼을 선택합니다.
lowerbound, upperbound :
lowerBound 및 upperBound는 파티션 보폭을 결정하는 데만 사용되며 테이블의 행을 필터링하는 데 사용되지 않습니다. 따라서 테이블의 모든 행이 분할되어 반환됩니다. ( 소스 ) 아래에서 Spark가 각 파티션을 생성하는 방법을 볼 수 있습니다. 이것이 낮은 카디널리티 열을 파티션 열로 원하지 않는 이유이지만 지금은 이해가 되지 않는 경우를 대비하여 아래에 예를 들어 보겠습니다.

luminousmen.com의 luminousmen 사진

분할에 사용하려는 열에 값이 0과 1뿐이라고 상상해 보십시오. 극단적인 예이지만 부울 값 True 및 False 가 숫자 값으로 변환되는 데이터 엔지니어링의 일반적인 시나리오입니다.

0과 1만 있기 때문에 0을 하한 으로 , 1을 상한으로 사용합니다. 파티션이 생성되는 방식에 따라(위 그림 참조) 값이 0인 모든 행이 첫 번째 파티션으로 푸시되고 (첫 번째 파티션의 WHERE 절은 0이 필터링되지 않는 유일한 절임) 모든 행이 값이 1이면 마지막 파티션으로 푸시됩니다( 마지막 파티션의 WHERE 절은 1이 필터링되지 않는 유일한 절입니다) .

따라서 생성한 파티션 수에 관계없이 두 개의 파티션만 채워지고 두 개의 실행기 코어만 모든 작업을 수행하며 다른 것들은 예약된 다른 작업이 없는 경우 유휴 상태가 됩니다. 이로 인해 쿼리 성능이 저하됩니다.

데이터 왜도

20M 행과 30개의 파티션, 하한 및 상한이 2020–01–01 및 2022–12–31인 데이터 세트가 있다고 가정해 보겠습니다. 이 기사의 목적을 위해 30개의 실행기 코어가 있는 Spark 세션을 사용하겠습니다(동시에 최대 30개의 작업을 실행할 수 있음을 의미). Spark는 파티션당 하나의 작업을 할당하므로 각 파티션은 하나의 실행기 코어에서 처리됩니다.

1년 분량의 데이터를 10개의 파티션으로 나누고 1개의 파티션에 1개의 작업이 할당되는 것을 고려하면 2020년과 2021년에 할당된 각 작업은 100k 행을 처리해야 하지만 2022년에 할당된 각 작업은 처리해야 합니다. 180만 행. 이는 하나의 작업으로 처리해야 하는 데이터가 1700% 증가한 것입니다.

그 결과 처음 20개의 작업은 할당된 파티션 처리를 즉시 완료하고 완료된 작업에 할당된 20개의 실행기 코어는 마지막 작업까지 유휴 상태로 있습니다(유휴 실행기 코어를 사용할 수 있는 다른 작업이 없는 경우). 10개의 작업이 1700% 더 많은 레코드 처리를 완료합니다. 이는 특히 EMR 클러스터, Glue Job 또는 Databricks에서 Spark를 사용하는 경우 리소스 낭비 및 운영 비용 증가입니다.

어떻게 고칠 수 있습니까? 가장 간단한 솔루션: 파티셔닝에 다른 열을 사용하는 것입니다. 하지만 이 열을 사용하는 것이 중요한 경우 하나가 아닌 두 개의 쿼리와 두 개의 데이터 프레임을 사용할 수 있습니다. 왜곡된 데이터 다운스트림 문제(병렬 계산 시스템에서 반복되는 문제)가 여전히 있지만 한 번에 한 가지 문제를 해결합니다.

첫 번째 쿼리는 처음 2년을 로드하고(30개의 실행기 코어는 2020년과 2021년 사이에 분할됨) 두 번째 쿼리는 마지막 연도를 로드합니다(따라서 이제 30개의 실행기 코어가 모두 2022년에 사용됨). 그로 인해 2022년과 관련된 주어진 작업은 이제 180만 행이 아닌 60만 행을 처리해야 합니다. 이는 원래 구현에 비해 단일 실행기 코어에서 처리해야 하는 행이 66% 감소한 것이며, 그 위에 이제 유휴 상태로 두는 대신 Spark 세션의 모든 실행기 코어를 사용합니다.

숫자, 날짜 또는 타임스탬프 열이 없거나 열에 왜곡된 데이터가 있으면 어떻게 합니까?

문제가 아니다. 데이터 엔지니어로서 CTE 및 창 함수를 사용하여 숫자 열을 생성하는 것은 row_number()식은 죽 먹기여야 합니다. 새 열이 생성될 때 약간의 오버헤드가 발생하지만 이 오버헤드가 있더라도 하나의 실행기 코어를 사용하여 하나의 파티션에 모든 데이터를 로드하려고 시도하는 것보다 새로 생성된 열을 사용하여 최종 쿼리가 훨씬 빠르게 실행됩니다.

with cte as
(
    SELECT f1, f2, f3
    FROM table
)
select *, row_number() over (order by f1) as rn from cte

original_query = "SELECT f1, f2, f3 FROM table"

final_query = f'(with cte as ({original_query}) select *, row_number() over (order by f1) as rn from cte) as q'

하나가 아닌 두 개의 쿼리를 실행합니다. - 첫 번째 쿼리는 partitionColumn
으로 사용할 열의 최소값과 최대값을 찾기 위해 실행됩니다. - 두 번째 쿼리에서는 첫 번째 쿼리의 결과를 하한 및 상한 인수 로 전달합니다. .