EMR 클러스터에서 Spark를 사용하여 Redshift 클러스터에 연결하려면 어떻게 해야 하나요?

해결 방법

참고: 다음 단계를 진행하기에 앞서 Redshift 클러스터와 EMR 클러스터를 구성하고 Spark 서비스를 설치합니다.

EMR 클러스터에서 Redshift 클러스터로의 연결 테스트

1.    EMR 기본, 코어 및 태스크 노드 보안 그룹이 TCP 포트 5439에 대한 Redshift의 보안 그룹(인바운드 규칙)에서 허용되는지 확인합니다. EMR 및 Redshift 클러스터가 서로 다른 두 개의 Amazon Virtual Private Cloud(Amazon VPC)에 배포된 경우 VPC 피어링을 구성합니다.

2.    SSH를 사용하여 EMR 프라이머리 노드에 연결하고 다음 Telnet 명령을 실행합니다. 이 Telnet 명령은 EMR 클러스터와 Redshift 클러스터 간에 연결을 설정할 수 있는지 확인해 줍니다. 다음 명령에서 Redshift_Endpoint를 Redshift 클러스터의 올바른 엔드포인트로 바꿉니다.

telnet Redshift_Endpoint 5439

다음은 성공적인 연결에 대한 예제 출력입니다.

telnet redshift-cluster-1.XXXX.us-east-1.redshift.amazonaws.com 5439
Trying 172.31.48.21...
Connected to redshift-cluster-1.XXXXX.us-east-1.redshift.amazonaws.com.
Escape character is

EMR-5.x.x 시리즈 클러스터에서 Spark를 사용하여 Redshift 클러스터에 연결

Databrick의 Spark-Redshift 패키지(라이브러리)를 사용합니다. 이 라이브러리는 Amazon Redshift에서 Spark SQL DataFrames로 데이터를 로드하고 DataFrames를 다시 Amazon Redshift 테이블에 저장합니다.

1.    SSH를 사용하여 EMR 프라이머리 노드에 연결합니다.

2.    Spark-Redshift 라이브러리에서 작업하려면 다음의 .jar 파일을 EMR 클러스터로 다운로드합니다.

wget https://repo1.maven.org/maven2/com/databricks/spark-redshift_2.11/2.0.1/spark-redshift_2.11-2.0.1.jar
wget https://github.com/ralfstx/minimal-json/releases/download/0.9.4/minimal-json-0.9.4.jar

3.    다운로드한 JAR 파일을 기본 Spark 라이브러리에 복사합니다. Spark 라이브러리 경로는 **/usr/lib/spark/jars/**입니다.

sudo cp spark-redshift_2.11-2.0.1.jar /usr/lib/spark/jars/
sudo cp minimal-json-0.9.4.jar /usr/lib/spark/jars/

4.    Amazon Redshift JDBC 드라이버를 사용하여 spark-shell 명령을 실행하여 Redshift 클러스터에 연결합니다. JDBC 드라이버는 Amazon EMR 버전 4.7.0 이상에 포함되어 있습니다.

spark-shell --jars /usr/share/aws/redshift/jdbc/RedshiftJDBC41.jar

5.    spark-shell 세션이 초기화되면 다음 단계를 실행하여 Redshift 클러스터에 연결합니다. 다음 명령에서 사용 사례에 따라 Amazon Redshift 엔드포인트, Amazon Simple Storage Service(Amazon S3) 버킷 이름 및 테이블 세부 정보를 업데이트합니다.

import com.amazonaws.auth.AWSCredentialsProvider
import com.amazonaws.auth.AWSSessionCredentials
import com.amazonaws.auth.InstanceProfileCredentialsProvider

// Instance Profile for authentication to AWS resources
val provider = new InstanceProfileCredentialsProvider();
val credentials: AWSSessionCredentials = provider.getCredentials.asInstanceOf[AWSSessionCredentials];
val token = credentials.getSessionToken;
val awsAccessKey = credentials.getAWSAccessKeyId;
val awsSecretKey = credentials.getAWSSecretKey

// Set JDBC URL of Redshift
val jdbcUrl = "jdbc:redshift://<cluster-name>.<id>.<region>.redshift.amazonaws.com:5439/<database>?user=<user>&password=<password>"

// Create DataFrame by loading Redshift query
val df = spark.read.format("com.databricks.spark.redshift").option("url", jdbcUrl).option("tempdir", "s3://<S3-path-to-store-temp-data>").option("query", "select * from <table-name>").option("temporary_aws_access_key_id", awsAccessKey).option("temporary_aws_secret_access_key", awsSecretKey).option("temporary_aws_session_token", token).load()
df.show(2)

Amazon EMR-6.x.x 시리즈 클러스터에서 Spark를 사용하여 Redshift 클러스터에 연결

Amazon EMR 버전 6.x 이상은 Scala 버전 2.12를 사용합니다. Amazon EMR 5.x는 Scala 버전 2.11을 사용합니다. Amazon EMR 5.x 버전에서 사용하는 spark-redshift_2.11-2.0.1.jar 파일은 Amazon EMR 버전 6.x 이상과 호환되지 않습니다. 따라서 Amazon EMR 6.x 이상의 클러스터에서는 spark-redshift_2.12-4.2.0.jar 커넥터를 사용하세요.

1.    SSH를 사용하여 EMR 프라이머리 노드에 연결합니다.

2.    Spark-Redshift 라이브러리에서 작업하려면 다음의 .jar 파일을 EMR 클러스터로 다운로드합니다.

wget https://repo1.maven.org/maven2/io/github/spark-redshift-community/spark-redshift_2.12/4.2.0/spark-redshift_2.12-4.2.0.jar
wget https://github.com/ralfstx/minimal-json/releases/download/0.9.4/minimal-json-0.9.4.jar

3.    다운로드한 JAR 파일을 기본 Spark 라이브러리에 복사합니다. Spark 라이브러리 경로는**/usr/lib/spark/jars/**입니다.

sudo cp spark-redshift_2.12-4.2.0.jar /usr/lib/spark/jars/
sudo cp minimal-json-0.9.4.jar /usr/lib/spark/jars/

4.    Amazon Redshift JDBC 드라이버를 사용하여 spark-shell 명령을 실행해서 Redshift 클러스터에 연결합니다. JDBC 드라이버는 EMR 버전 4.7.0 이상에 포함되어 있습니다.

spark-shell --jars /usr/share/aws/redshift/jdbc/RedshiftJDBC41.jar

5.    spark-shell 세션이 초기화되면 다음 단계를 실행하여 Redshift 클러스터에 연결합니다. 다음 명령에서 사용 사례에 따라 Amazon Redshift 엔드포인트, S3 버킷 이름 및 테이블 세부 정보를 업데이트합니다.

import com.amazonaws.auth.AWSCredentialsProvider
import com.amazonaws.auth.AWSSessionCredentials
import com.amazonaws.auth.InstanceProfileCredentialsProvider

// Instance Profile for authentication to AWS resources
val provider = new InstanceProfileCredentialsProvider();
val credentials: AWSSessionCredentials = provider.getCredentials.asInstanceOf[AWSSessionCredentials];
val token = credentials.getSessionToken;
val awsAccessKey = credentials.getAWSAccessKeyId;
val awsSecretKey = credentials.getAWSSecretKey

// Set JDBC URL of Redshift
val jdbcUrl = "jdbc:redshift://<cluster-name>.<id>.<region>.redshift.amazonaws.com:5439/<database>?user=<user>&password=<password>"

// Create DataFrame by loading Redshift query
val df = spark.read.format("io.github.spark_redshift_community.spark.redshift").option("url", jdbcUrl).option("tempdir", "s3://bucket/tmp/").option("query", "select * from <table>").option("temporary_aws_access_key_id", awsAccessKey).option("temporary_aws_secret_access_key", awsSecretKey).option("temporary_aws_session_token", token).load()
df.show(2)

---