使用适用于 Apache Spark 的 Amazon Redshift 集成进行身份验证

使用 AWS Secrets Manager 检索凭证并连接到 Amazon Redshift

以下代码示例显示了如何使用 AWS Secrets Manager 检索凭证，以通过 Python 中 Apache Spark 的 PySpark 接口连接到 Amazon Redshift 集群。

from pyspark.sql import SQLContext
import boto3

sc = # existing SparkContext
sql_context = SQLContext(sc)

secretsmanager_client = boto3.client('secretsmanager')
secret_manager_response = secretsmanager_client.get_secret_value(
    SecretId='string',
    VersionId='string',
    VersionStage='string'
)
username = # get username from secret_manager_response
password = # get password from secret_manager_response
url = "jdbc:redshift://redshifthost:5439/database?user=" + username + "&password=" + password

# Read data from a table
df = sql_context.read \
    .format("io.github.spark_redshift_community.spark.redshift") \
    .option("url", url) \
    .option("dbtable", "my_table") \
    .option("tempdir", "s3://path/for/temp/data") \
    .load()

使用 IAM 检索凭证并连接到 Amazon Redshift

您可以使用 Amazon Redshift 提供的 JDBC 版本 2 驱动程序，通过 Spark 连接器连接到 Amazon Redshift。要使用 AWS Identity and Access Management（IAM），请将您的 JDBC URL 配置为使用 IAM 身份验证。要从 Amazon EMR 连接到 Redshift 集群，您必须授予 IAM 角色权限以检索临时 IAM 凭证。将以下权限分配到 IAM 角色，使其能够检索凭证并运行 Amazon S3 操作。

• Redshift:GetClusterCredentials（适用于预置的 Redshift 集群）

• Redshift:DescribeClusters（适用于预置的 Redshift 集群）

• Redshift:GetWorkgroup（适用于 Amazon Redshift Serverless 工作组）

• Redshift:GetCredentials（适用于 Amazon Redshift Serverless 工作组）

• s3:GetBucket

• s3:GetBucketLocation

• s3:GetObject

• s3:PutObject

• s3:GetBucketLifecycleConfiguration

有关 GetClusterCredentials 的更多信息，请参阅 GetClusterCredentials 的资源策略。

您还必须确保 Amazon Redshift 可以在 COPY 和 UNLOAD 操作期间担任 IAM 角色。

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "redshift.amazonaws.com"
            },
            "Action": "sts:AssumeRole"
        }
    ]
}

以下示例在 Spark 和 Amazon Redshift 之间使用 IAM 身份验证：

from pyspark.sql import SQLContext
import boto3

sc = # existing SparkContext
sql_context = SQLContext(sc)

url = "jdbc:redshift:iam//redshift-host:redshift-port/db-name"
iam_role_arn = "arn:aws:iam::account-id:role/role-name"

# Read data from a table
df = sql_context.read \
    .format("io.github.spark_redshift_community.spark.redshift") \
    .option("url", url) \
    .option("aws_iam_role", iam_role_arn) \
    .option("dbtable", "my_table") \
    .option("tempdir", "s3a://path/for/temp/data") \
    .mode("error") \
    .load()