Spark - 병렬 JDBC Table 읽기

 

Spark에서 PostgreSQL 테이블을 읽는 것을 예제로 합니다.

직렬 읽기는 단일 커넥션을 통해 데이터가 순차적으로 로드되며,

병렬 읽기는 여러 분할을 통해 데이터를 동시에 읽음으로써 성능을 크게 향상시킬 수 있습니다.

 

병렬 읽기가 무조건 좋은 것 처럼 보이지만

클러스터의 리소스나 테이블의 크기에 따라 병렬 읽기가 필요하지 않은 경우도 있습니다.

 

1. 직렬 읽기 (Sequential Read)

직렬 읽기는 단일 커넥션으로 PostgreSQL 테이블의 데이터를 Spark로 읽어옵니다. 

Spark는 이 커넥션을 통해 데이터를 가져오고, 모든 데이터를 하나의 파티션으로 로드합니다.

예제 코드 (pysark)

from pyspark.sql import SparkSession

# SparkSession 생성
spark = SparkSession.builder.appName("PostgresSerialRead").getOrCreate()

# PostgreSQL 테이블 직렬 읽기
df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://your-postgres-url:5432/your_database") \
    .option("dbtable", "your_table") \
    .option("user", "your_username") \
    .option("password", "your_password") \
    .load()

 

예제 코드 (databricks)

CREATE TABLE IF NOT EXISTS twitterJDBC
USING org.apache.spark.sql.jdbc
OPTIONS (
  driver "org.postgresql.Driver",
  url "jdbc:postgresql://server1.databricks.training:5432/training",
  user "readonly",
  password "readonly",
  dbtable "training.Account"
  --조건절 푸쉬다운 사용시
  --dbtable "(SELECT * FROM training.Account WHERE location = 'Philippines') as subq"
)

• 단일 파티션으로 데이터를 읽습니다.
• 데이터 양이 적을 때 적합합니다.
• 모든 데이터가 하나의 파티션으로 로드되어 속도가 느릴 수 있습니다.
• 큰 테이블을 읽어올 때 성능이 떨어질 수 있습니다.

 

2. 병렬 읽기 (Parallel Read)

병렬 읽기는 여러 커넥션을 통해 데이터를 동시에 읽어오는 방식으로, 

특정 열을 기준으로 데이터 파티션을 나누어 병렬로 로드합니다. 

PostgreSQL에서 분할 키(주로 숫자 열이나 고유 ID)로 데이터를 나누고, 

여러 Spark 태스크가 이를 병렬로 가져옵니다.

예제 코드 (pyspark)

from pyspark.sql import SparkSession

# SparkSession 생성
spark = SparkSession.builder.appName("PostgresParallelRead").getOrCreate()

# PostgreSQL 테이블 병렬 읽기
df = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql://your-postgres-url:5432/your_database") \
    .option("dbtable", "your_table") \
    .option("user", "your_username") \
    .option("password", "your_password") \
    .option("partitionColumn", "id") \  # 분할 기준 컬럼 (주로 숫자 또는 고유 ID)
    .option("lowerBound", "1") \        # 분할 시작 값
    .option("upperBound", "10000") \    # 분할 끝 값
    .option("numPartitions", "10") \    # 파티션 수 (병렬로 읽을 작업 수)
    .load()

 

예제 코드 (databricks)

CREATE TABLE IF NOT EXISTS twitterParallelJDBC
USING org.apache.spark.sql.jdbc
OPTIONS (
  driver "org.postgresql.Driver",
  url "jdbc:postgresql://server1.databricks.training:5432/training",
  user "readonly",
  password "readonly",
  dbtable "training.Account",
  partitionColumn '"userID"',
  lowerBound 2591,
  upperBound 951253910555168768,
  numPartitions 25
)

• 여러 파티션으로 데이터를 나누어 병렬로 읽기 때문에 속도가 빠릅니다.
• 데이터 양이 많을 때 적합합니다.
• 데이터 파티션이 균등하지 않으면 성능이 떨어질 수 있습니다.
• 순서를 지정할 수 없는 범주형 컬럼(예시: 학급명 - 개나리반, 진달래반 ..)일 때 lowerBound, upperBound를 사용할 수 없습니다.
• 범주형 컬럼 값을 때는 numPartitions로 학급 수 만큼 지정합니다.

 

결론

작은 데이터는 직렬 읽기로 충분히 처리할 수 있으며, 설정이 간단합니다.

큰 데이터는 병렬 읽기를 통해 속도를 높일 수 있습니다. 

병렬 읽기를 사용할 경우, 분할 키와 파티션 수 설정을 적절히 해야 효율적인 데이터 로드가 가능합니다.

 

실행속도 차이