Como faço para me conectar a um cluster do Redshift usando o Spark no meu cluster do EMR?

Resolução

**Observação:**configure seu cluster do Redshift e o cluster do EMR e instale o serviço Spark antes de prosseguir com as etapas a seguir.

Teste a conectividade do cluster do EMR com o cluster do Redshift

1.    Verifique se os grupos de segurança primários, principais e de nós de tarefas do EMR são permitidos no grupo de segurança do Redshift (regras de entrada) para a porta TCP 5439. Se os clusters do EMR e do Redshift forem implantados em duas Amazon Virtual Private Clouds (Amazon VPC) diferentes, configure o emparelhamento de VPC.

2.    Conecte-se ao nó primário do EMR usando SSH e execute o comando Telnet a seguir. Esse comando Telnet verifica se você pode estabelecer uma conexão entre o cluster do EMR e o cluster do Redshift. No comando a seguir, substitua Redshift_Endpoint pelo endpoint correto para seu cluster do Redshift.

telnet Redshift_Endpoint 5439

Veja a seguir um exemplo de saída para uma conexão bem-sucedida:

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

Conecte-se ao cluster do Redshift usando o Spark nos clusters da série EMR-5.x.x

Use o pacote spark-redshift (biblioteca) do Databrick. Essa biblioteca carrega dados no Spark SQL DataFrames do Amazon Redshift e alva os DataFrames de volta nas tabelas do Amazon Redshift.

1.    Conecte-se ao nó primário do EMR usando SSH.

2.    Para trabalhar com a biblioteca spark-redshift, baixe os seguintes arquivos .jar no cluster do 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 os arquivos JAR baixados para a biblioteca padrão do Spark. O caminho para biblioteca do 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.    Execute o comando spark-shell usando o driver JDBC do Amazon Redshift para se conectar ao cluster do Redshift. O driver JDBC está incluído nas versões 4.7.0 e superiores do Amazon EMR.

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

5.    Depois que a sessão spark-shell for inicializada, execute as etapas a seguir para se conectar ao cluster do Redshift. Nos comandos a seguir, atualize o endpoint do Amazon Redshift, o nome do bucket do Amazon Simple Storage Service (Amazon S3) e os detalhes da tabela de acordo com seu 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)

Conecte-se ao cluster do Redshift usando o Spark nos clusters da série Amazon EMR-6.x.x

As versões 6.x e superiores do Amazon EMR usam o Scala versão 2.12. O Amazon EMR 5.x usa o Scala versão 2.11. O arquivo spark-redshift_2.11-2.0.1.jar usado pela versão do Amazon EMR 5.x não é compatível com o Amazon EMR versão 6.x e superior. Portanto, use o conector spark-redshift_2.12-4.2.0.jar nos clusters do Amazon EMR 6.x e superiores.

1.    Conecte-se ao nó primário do EMR usando SSH.

2.    Para trabalhar com a biblioteca spark-redshift, baixe os seguintes arquivos .jar no cluster do 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 os arquivos JAR baixados para a biblioteca padrão do Spark. O caminho da biblioteca do 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.    Execute o comando spark-shell usando o driver JDBC do Amazon Redshift para se conectar ao cluster do Redshift. O driver JDBC está incluído nas versões 4.7.0 do EMR e superiores.

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

5.    Depois que a sessão spark-shell for inicializada, execute as etapas a seguir para se conectar ao cluster do Redshift. Nos comandos a seguir, atualize o endpoint do Amazon Redshift, o nome do bucket do S3 e os detalhes da tabela de acordo com seu 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)

---