In che modo posso connettermi a un cluster Redshift utilizzando Spark nel cluster EMR?

Risoluzione

Nota: configura il cluster Redshift e il cluster EMR e installa il servizio Spark prima di procedere con i passaggi seguenti.

Verifica la connettività tra il cluster EMR e il cluster Redshift

1.    Verifica che i gruppi di sicurezza dei nodi primari, principali e attività EMR siano consentiti nel gruppo di sicurezza di Redshift (regole in entrata) per la porta TCP 5439. Se i cluster EMR e Redshift sono implementati in due diversi Amazon Virtual Private Cloud (Amazon VPC), configura il peering VPC.

2.    Connettiti al nodo primario EMR tramite SSH ed esegui il seguente comando Telnet. Il comando Telnet verifica che sia possibile stabilire una connessione tra il cluster EMR e il cluster Redshift. Nel comando seguente, sostituisci Redshift_Endpoint con l'endpoint corretto per il tuo cluster Redshift.

telnet Redshift_Endpoint 5439

Di seguito è riportato un output di esempio per una connessione riuscita:

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

Connessione al cluster Redshift tramite Spark nei cluster della serie EMR-5.x.x

Usa il pacchetto (libreria) Spark-Redshift di Databrick. Questa libreria carica i dati nei DataFrame di Spark SQL da Amazon Redshift e salva nuovamente i DataFrame nelle tabelle Amazon Redshift.

1.    Connettiti al nodo primario EMR tramite SSH.

2.    Per lavorare con la libreria Spark-Redshift, scarica i seguenti file .jar nel 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.    Copia i file JAR scaricati nella libreria Spark predefinita. Il percorso della libreria 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.    Esegui il comando spark-shell utilizzando il driver JDBC di Amazon Redshift per connetterti al cluster Redshift. Il driver JDBC è incluso in Amazon EMR 4.7.0 e versioni successive.

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

5.    Dopo l'inizializzazione della sessione spark-shell, procedi come segue per connetterti al cluster Redshift. Nei comandi seguenti, aggiorna l'endpoint Amazon Redshift, il nome del bucket Amazon Simple Storage Service (Amazon S3) e i dettagli della tabella in base al caso d'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)

Connessione al cluster Redshift tramite Spark nei cluster della serie Amazon EMR-6.x.x

Amazon EMR 6.x e versioni successive utilizza Scala versione 2.12. Amazon EMR 5.x utilizza Scala versione 2.11. Il file spark-redshift_2.11-2.0.1.jar utilizzato dalla versione Amazon EMR 5.x non è compatibile con Amazon EMR 6.x e versioni successive. Quindi, usa il connettore spark-redshift_2.12-4.2.0.jar nei cluster Amazon EMR 6.x e versioni successive.

1.    Connettiti al nodo primario EMR tramite SSH.

2.    Per lavorare con la libreria Spark-Redshift, scarica i seguenti file .jar nel 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.    Copia i file JAR scaricati nella libreria Spark predefinita. Il percorso della libreria 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.    Esegui il comando spark-shell utilizzando il driver JDBC di Amazon Redshift per connetterti al cluster Redshift. Il driver JDBC è incluso in EMR 4.7.0 e versioni successive.

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

5.    Dopo l'inizializzazione della sessione spark-shell, procedi come segue per connetterti al cluster Redshift. Nei comandi seguenti, aggiorna l'endpoint Amazon Redshift, il nome del bucket S3 e i dettagli della tabella in base al caso d'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)

---