YARN Cluster에 연결된 Spark로 pyspark - Oracle 데이터베이스 연결하기

Step 1. OJDBC 드라이버 설정

https://www.oracle.com/kr/database/technologies/appdev/jdbc.html

JDBC 드라이버 | Oracle 대한민국

• JDBC란?
  • Java Database Connectivity의 약자로, 자바에서 데이터베이스에 접속할 수 있도록 하는 자바 API
  • 일관된 방식으로 데이터베이스에 접근할 수 있게 해주므로 구체적인 데이터베이스의 시스템 세부사항을 걱정하지 않고도 작업을 수행할 수 있다.
• Pyspark, JVM 등 적절한 버전을 찾아 다운로드한다.
• ojdbc.jar 는 spark의 각각 driver, executor, yarn 클러스터가 읽을 수 있는 경로에 위치해야 하는데,
  • Deploy Mode가 Local인 경우 jdbc는 local머신에 위치하여 spark의 driver, executor가 읽을 수 있는 곳에 위치해야 하고
  • Deploy Mode가 Client인 경우 driver는 클라이언트에, executor는 클러스터에 위치하고
  • Deploy Mode가 Cluster인 경우 driver, executor가 모두 클러스터에 위치해야 한다.

 

Step 2. Spark Session 설정

• Pyspark 스크립트 내에서 Spark Session을 설정하자
• 여기에서 spark.driver.extraClassPath와 spark.executor.extraClassPath, 그리고 YARN 클러스터를 사용한다면 spark.yarn.jars에서 ojdbc 드라이버의 위치를 지정해줘야 한다.
• 아래 예시에서는 client mode이기 때문에 spark의 driver와 executor의 ojdbc 경로 위치가 다르다.
• 또한 YARN 클러스터를 사용하고 있는 경우 모든 executor들이 접근할 수 있는 위치에 ojdbc가 위치해주어야하고 그것은 HDFS에 위치해있기 때문에 HDFS 경로를 넣어주었다.

from pyspark import SparkConf, SparkContext, StorageLevel
from pyspark.sql import SparkSession, SQLContext

# ojdbc 경로 위치
ojdbc_local_path = "/path/to/ojdbc8.jar"
ojdbc_cluster_path = "hdfs://path/to/ojdbc8.jar"

# spark - YARN 클러스터 셋팅
appName = "Test"
spark = SparkSession.builder\
                    .appName(appName)\
                    .config("spark.jars", ojdbc_local_path)\
                    .config("spark.yarn.jars", ojdbc_cluster_path)\
                    .config("spark.driver.extraClassPath", ojdbc_local_path)\
                    .config("spark.executor.extraClassPath", ojdbc_cluster_path)

sc = spark.sparkContext

 

Step 3. JDBC 연결 설정

• JDBC를 통해 연결하고자 하는 데이터베이스의 정보를 입력해주어야 한다.
• 데이터베이스 서버의 IP, Port, DB명, ID(useR), password 등의 정보가 필요하다.

# Oracle 연결 정보
ip = ""
port = ""
service_name = ""
user = ""
password = ""

# spark - oracle 연결 정보
driver_format = "jdbc"
driver = "oracle.jdbc.OracleDriver"
url = f"jdbc:oracle:thin:@//{ip}:{port}/{service_name}"
fetchsize = 1000
query = "select * from TABLE_NM"

# 데이터 읽기
df = spark.read.format(driver_format)\
                .option("url", url)\
                .option("driver", driver)\
                .option("user", user)\
                .option("password", password)\
                .option("query", query)\
                .option("fetchsize", fetchsize)\
                .load()

• 여기에서 spark의 driver와 모든 executor 노드들이 데이터베이스 서버와 방화벽이 해제되어있어야 정상적으로 connection이 된다.