Comment me connecter à un cluster Redshift à l'aide de Spark dans mon cluster EMR ?

Solution

Remarque : configurez votre cluster Redshift et votre cluster EMR et installez le service Spark avant de procéder aux étapes suivantes.

Tester la connectivité entre le cluster EMR et le cluster Redshift

1.    Vérifiez que les groupes de sécurité EMR de nœud primaire, de base et de tâche sont autorisés dans le groupe de sécurité de Redshift (règles entrantes) pour le port TCP 5439. Si les clusters EMR et Redshift sont déployés dans deux Amazon Virtual Private Clouds (Amazon VPC) différents, configurez l'appairage de VPC.

2.    Connectez-vous au nœud primaire EMR via SSH et exécutez la commande Telnet suivante. Cette commande Telnet vérifie que vous pouvez établir une connexion entre le cluster EMR et le cluster Redshift. Dans la commande suivante, remplacez Redshift_Endpoint par le point de terminaison approprié pour votre cluster Redshift.

telnet Redshift_Endpoint 5439

Voici un exemple de sortie pour une connexion réussie :

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

Se connecter au cluster Redshift à l'aide de Spark dans les clusters de la série EMR-5.x.x

Utilisez le package spark-redshift (bibliothèque) de Databrick. Cette bibliothèque charge des données dans Spark SQL DataFrames depuis Amazon Redshift et enregistre également les DataFrames dans des tables Amazon Redshift.

1.    Connectez-vous au nœud primaire EMR via SSH.

2.    Pour utiliser la bibliothèque spark-redshift, téléchargez les fichiers .jar suivants dans le cluster 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.    Copiez les fichiers JAR téléchargés dans la bibliothèque Spark par défaut. Le chemin de la bibliothèque Spark est /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.    Exécutez la commande spark-shell à l'aide du pilote JDBC Amazon Redshift pour vous connecter au cluster Redshift. Le pilote JDBC est inclus dans les versions 4.7.0 et supérieures d'Amazon EMR.

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

5.    Une fois la session spark-shell initialisée, exécutez les étapes suivantes pour vous connecter au cluster Redshift. Dans les commandes suivantes, mettez à jour le point de terminaison Amazon Redshift, le nom du compartiment Amazon Simple Storage Service (Amazon S3) et les détails de la table en fonction de votre cas d'utilisation.

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)

Se connecter au cluster Redshift à l'aide de Spark dans les clusters de la série Amazon EMR-6.x.x

Les versions 6.x et supérieures d'Amazon EMR utilisent la version 2.12 de Scala. Amazon EMR 5.x utilise la version 2.11 de Scala. Le fichier spark-redshift_2.11-2.0.1.jar utilisé par la version 5.x d'Amazon EMR n'est pas compatible avec les versions 6.x et supérieures d'Amazon EMR. Par conséquent, utilisez le connecteur spark-redshift_2.12-4.2.0.jar dans les clusters Amazon EMR 6.x et versions ultérieures.

1.    Connectez-vous au nœud primaire EMR via SSH.

2.    Pour utiliser la bibliothèque spark-redshift, téléchargez les fichiers .jar suivants dans le cluster 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.    Copiez les fichiers JAR téléchargés dans la bibliothèque Spark par défaut. Le chemin de la bibliothèque Spark est /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.    Exécutez la commande spark-shell à l'aide du pilote JDBC Amazon Redshift pour vous connecter au cluster Redshift. Le pilote JDBC est inclus dans les versions 4.7.0 et supérieures d'EMR.

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

5.    Une fois la session spark-shell initialisée, exécutez les étapes suivantes pour vous connecter au cluster Redshift. Dans les commandes suivantes, mettez à jour le point de terminaison Amazon Redshift, le nom du compartiment S3 et les détails de la table en fonction de votre cas d'utilisation.

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)

---