¿Cómo me conecto a un clúster de Redshift mediante Spark en mi clúster de EMR?

Resolución

Nota: Configure el clúster de Redshift y el clúster de EMR, e instale el servicio de Spark antes de continuar con los siguientes pasos.

Prueba de la conectividad del clúster de EMR al clúster de Redshift

1.    Compruebe que los grupos de seguridad primarios, principales y de nodos de tareas de EMR estén permitidos en el grupo de seguridad de Redshift (reglas de entrada) para el puerto TCP 5439. Si los clústeres de EMR y Redshift se implementan en dos Amazon Virtual Private Cloud (Amazon VPC) diferentes, configure el emparejamiento de VPC.

2.    Conéctese al nodo principal de EMR mediante SSH y ejecute el siguiente comando Telnet. Este comando Telnet verifica que pueda establecer una conexión entre el clúster de EMR y el clúster de Redshift. En el siguiente comando, sustituya Redshift_Endpoint por el punto de conexión correcto para su clúster de Redshift.

telnet Redshift_Endpoint 5439

A continuación, se muestra un ejemplo de salida para una conexión correcta:

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

Conexión al clúster de Redshift mediante Spark en los clústeres de la serie EMR-5.x.x

Utilice el paquete spark-redshift de Databrick (biblioteca). Esta biblioteca carga datos en Spark SQL DataFrames desde Amazon Redshift y también vuelve a guardar DataFrames en tablas de Amazon Redshift.

1.    Conéctese al nodo principal de EMR mediante SSH.

2.    Para trabajar con la biblioteca spark-redshift, descargue los siguientes archivos .jar en el clúster de 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.    Copie los archivos JAR descargados a la biblioteca de Spark predeterminada. La ruta de la biblioteca de Spark es /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.    Ejecute el comando spark-shell con el controlador JDBC de Amazon Redshift para conectarse al clúster de Redshift. El controlador JDBC se incluye en las versiones 4.7.0 y posteriores de Amazon EMR.

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

5.    Una vez inicializada la sesión spark-shell, ejecute los siguientes pasos para conectarse al clúster de Redshift. En los siguientes comandos, actualice el punto de conexión de Amazon Redshift, el nombre del bucket de Amazon Simple Storage Service (Amazon S3) y los detalles de la tabla según su caso de uso.

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)

Conexión al clúster de Redshift mediante Spark en los clústeres de la serie Amazon EMR-6.x.x

Las versiones 6.x y posteriores de Amazon EMR utilizan la versión 2.12 de Scala. Amazon EMR 5.x usa la versión 2.11 de Scala. El archivo spark-redshift_2.11-2.0.1.jar que utiliza la versión 5.x de Amazon EMR no es compatible con la versión 6.x y posteriores de Amazon EMR. Por lo tanto, utilice el conector spark-redshift_2.12-4.2.0.jar en los clústeres de Amazon EMR 6.x y posteriores.

1.    Conéctese al nodo principal de EMR mediante SSH.

2.    Para trabajar con la biblioteca spark-redshift, descargue los siguientes archivos .jar en el clúster de 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.    Copie los archivos JAR descargados a la biblioteca de Spark predeterminada. La ruta de la biblioteca de Spark es /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.    Ejecute el comando spark-shell con el controlador JDBC de Amazon Redshift para conectarse al clúster de Redshift. El controlador JDBC se incluye en las versiones 4.7.0 y posteriores de EMR.

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

5.    Una vez inicializada la sesión spark-shell, ejecute los siguientes pasos para conectarse al clúster de Redshift. En los siguientes comandos, actualice el punto de conexión de Amazon Redshift, el nombre del bucket de S3 y los detalles de la tabla según su caso de uso.

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)

---