Amazon Redshift 將不再支援從修補程式 198 開始建立新的 Python UDFs。現有 Python UDF 將繼續正常運作至 2026 年 6 月 30 日。如需詳細資訊，請參閱[部落格文章](https://aws.amazon.com/blogs/big-data/amazon-redshift-python-user-defined-functions-will-reach-end-of-support-after-june-30-2026/)。 本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。 # Apache Spark 的 Amazon Redshift 整合 [Apache Spark](https://aws.amazon.com/emr/features/spark/) 是一種分散式處理架構和程式設計模型，可協助您執行機器學習、串流處理或圖形分析。與 Apache Hadoop 類似，Spark 是一種開放原始碼、分散式處理系統，通常用於大數據的工作負載。Spark 具有經過最佳化的有向無環圖 (DAG) 執行引擎，並且會主動快取記憶體中的資料。這可以提高效能，對於某些算法和互動式查詢來說更是如此。 這項整合可為您提供 Spark 連接器，讓您用於建置在 Amazon Redshift 和 Amazon Redshift Serverless 中讀取和寫入資料的 Apache Spark 應用程式。這些應用程式不會影響應用程式效能或資料的交易一致性。這項整合會自動納入到 [Amazon EMR](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/) 和 [AWS Glue](https://docs.aws.amazon.com/glue/latest/dg/)，因此您可以立即執行會在資料擷取和轉換管道中存取資料並將資料載入到 Amazon Redshift 的 Apache Spark 任務。 目前，您可以搭配此整合使用 Spark 的 3.3.x、3.4.x、3.5.x 和 4.0.0 版。 這項整合會提供下列功能： + AWS Identity and Access Management (IAM) 身分驗證。如需詳細資訊，請參閱 [Amazon Redshift 中的身分和存取管理](https://docs.aws.amazon.com/redshift/latest/mgmt/redshift-iam-authentication-access-control.html)。 + 述詞和查詢下推，以提升效能。 + Amazon Redshift 資料類型。 + 對 Amazon Redshift 和 Amazon Redshift Serverless 的連線。 ## 使用 Spark 連接器時的考量和限制 + tempdir URI 會指向 Amazon S3 位置。此臨時目錄不會自動清理，因此可能會增加額外的費用。建議您使用《Amazon Simple Storage Service 使用者指南》**中的 [Amazon S3 生命週期政策](https://docs.aws.amazon.com/AmazonS3/latest/userguide/object-lifecycle-mgmt.html)來定義 Amazon S3 儲存貯體的保留規則。 + 根據預設，如果 Amazon S3和 Redshift 叢集位於不同的 AWS 區域，Amazon S3 和 Redshift 之間的複本將無法運作。若要使用個別 AWS 區域，請將 `tempdir_region` 參數設定為用於 的 S3 儲存貯體區域`tempdir`。 + 如果使用 `tempformat` 參數寫入 Parquet 資料，則會在 S3 和 Redshift 之間進行跨區域寫入。 + 建議您使用 [Amazon S3 伺服器端加密](https://docs.aws.amazon.com/AmazonS3/latest/userguide/serv-side-encryption.html)來加密所用的 Amazon S3 儲存貯體。 + 建議您[封鎖對 Amazon S3 儲存貯體的公開存取](https://docs.aws.amazon.com/AmazonS3/latest/userguide/access-control-block-public-access.html)。 + 建議您不要將 Amazon Redshift 叢集設為可公開存取。 + 建議您開啟 [Amazon Redshift 稽核記錄](https://docs.aws.amazon.com/redshift/latest/mgmt/db-auditing.html)。 + 建議您開啟 [Amazon Redshift 靜態加密](https://docs.aws.amazon.com/redshift/latest/mgmt/security-server-side-encryption.html)。 + 建議您針對從 Amazon EMR 上的 Spark 到 Amazon Redshift 的 JDBC 連線開啟 SSL。 + 建議您使用 Amazon Redshift 身分驗證參數的 `aws_iam_role` 參數傳遞 IAM 角色。 # 使用 Spark 連接器進行身分驗證 下圖描述了 Amazon S3、Amazon Redshift、Spark 驅動程式和 Spark 執行程式之間的身分驗證。 ![\[這是 Spark 連接器的身分驗證圖。\]](http://docs.aws.amazon.com/zh_tw/redshift/latest/mgmt/images/spark-connector-authentication.png) ## Redshift 和 Spark 之間的身分驗證 您可以使用 Amazon Redshift 提供的 JDBC 驅動器 2.x 版，透過指定登入憑證使用 Spark 連接器連線到 Amazon Redshift。若要使用 IAM，請[設定 JDBC url 以使用 IAM 身分驗證](https://docs.aws.amazon.com/redshift/latest/mgmt/generating-iam-credentials-configure-jdbc-odbc.html)。若要從 Amazon EMR 或 連線至 Redshift 叢集 AWS Glue，請確定您的 IAM 角色具有擷取臨時 IAM 登入資料的必要許可。下列清單描述 IAM 角色要擷取憑證和執行 Amazon S3 操作所需的所有許可。 + [ Redshift:GetClusterCredentials](https://docs.aws.amazon.com/redshift/latest/APIReference/API_GetClusterCredentials.html) (適用於佈建的 Redshift 叢集) + [ Redshift:DescribeClusters](https://docs.aws.amazon.com/redshift/latest/APIReference/API_DescribeClusters.html) (適用於佈建的 Redshift 叢集) + [ Redshift:GetWorkgroup](https://docs.aws.amazon.com/redshift-serverless/latest/APIReference/API_GetWorkgroup.html) (適用於 Amazon Redshift Serverless 工作群組) + [ Redshift:GetCredentials](https://docs.aws.amazon.com/redshift-serverless/latest/APIReference/API_GetCredentials.html) (適用於 Amazon Redshift Serverless 工作群組) + [ s3:ListBucket](https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListBuckets.html) + [ s3:GetBucket](https://docs.aws.amazon.com/AmazonS3/latest/API/API_control_GetBucket.html) + [ s3:GetObject](https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetObject.html) + [ s3:PutObject](https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html) + [ s3:GetBucketLifecycleConfiguration](https://docs.aws.amazon.com/AmazonS3/latest/API/API_GetBucketLifecycleConfiguration.html) 如需 GetClusterCredentials 的詳細資訊，請參閱 [ GetClusterCredentials 的 IAM 政策](https://docs.aws.amazon.com/redshift/latest/mgmt/redshift-iam-access-control-identity-based.html#redshift-policy-resources.getclustercredentials-resources)。 您還必須確保 Amazon Redshift 可以在 `COPY` 和 `UNLOAD` 操作期間擔任 IAM 角色。 ------ #### [ JSON ] **** ``` { "Version":"2012-10-17", "Statement": [ { "Effect": "Allow", "Principal": { "Service": "redshift.amazonaws.com" }, "Action": "sts:AssumeRole" } ] } ``` ------ 如果您使用的是最新的 JDBC 驅動程式，該驅動程式會自動管理從 Amazon Redshift 自我簽署憑證到 ACM 憑證的轉換。但是，您必須[指定 JDBC url 的 SSL 選項](https://docs.aws.amazon.com/redshift/latest/mgmt/jdbc20-configuration-options.html#jdbc20-ssl-option)。 以下是如何指定 JDBC 驅動程式 URL 和 `aws_iam_role` 以連線到 Amazon Redshift 的範例。 ``` df.write \ .format("io.github.spark_redshift_community.spark.redshift ") \ .option("url", "jdbc:redshift:iam://") \ .option("dbtable", "") \ .option("tempdir", "s3a:///") \ .option("aws_iam_role", "") \ .mode("error") \ .save() ``` ## Amazon S3 和 Spark 之間的身分驗證 如果您使用 IAM 角色在 Spark 和 Amazon S3 之間進行驗證，請使用下列其中一種方法： + 適用於 Java 的 AWS SDK 會使用 DefaultAWSCredentialsProviderChain 類別實作的預設登入資料提供者鏈結，自動嘗試尋找 AWS 登入資料。如需詳細資訊，請參閱[使用預設憑證提供者鏈結](https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html#credentials-default)。 + 您可以透過 [ Hadoop 組態屬性](https://github.com/apache/hadoop/blob/trunk/hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/index.md)指定 AWS 金鑰。例如，如果您的 `tempdir` 組態指向 `s3n://` 檔案系統，請在 Hadoop XML 組態檔案中設定 `fs.s3n.awsAccessKeyId` 和 `fs.s3n.awsSecretAccessKey` 屬性，或呼叫 `sc.hadoopConfiguration.set()` 以變更 Spark 的全域 Hadoop 組態。 例如，假設您使用 s3n 檔案系統，請新增： ``` sc.hadoopConfiguration.set("fs.s3n.awsAccessKeyId", "YOUR_KEY_ID") sc.hadoopConfiguration.set("fs.s3n.awsSecretAccessKey", "YOUR_SECRET_ACCESS_KEY") ``` 若為 s3a 檔案系統，請新增： ``` sc.hadoopConfiguration.set("fs.s3a.access.key", "YOUR_KEY_ID") sc.hadoopConfiguration.set("fs.s3a.secret.key", "YOUR_SECRET_ACCESS_KEY") ``` 如果您使用 Python，請使用下列操作： ``` sc._jsc.hadoopConfiguration().set("fs.s3n.awsAccessKeyId", "YOUR_KEY_ID") sc._jsc.hadoopConfiguration().set("fs.s3n.awsSecretAccessKey", "YOUR_SECRET_ACCESS_KEY") ``` + 在 `tempdir` URL 中編碼身分驗證金鑰。例如，URI `s3n://ACCESSKEY:SECRETKEY@bucket/path/to/temp/dir` 會編碼金鑰對 (`ACCESSKEY`,`SECRETKEY`)。 ## Redshift 和 Amazon S3 之間的身分驗證 如果您在查詢中使用 COPY 和 UNLOAD 命令，則也必須向 Amazon S3 授予 Amazon Redshift 的存取權，以代表您執行查詢。若要這麼做，請先[授權 Amazon Redshift 存取其他 AWS 服務](https://docs.aws.amazon.com/redshift/latest/mgmt/authorizing-redshift-service.html)，然後使用 [IAM 角色授權 COPY 和 UNLOAD 操作](https://docs.aws.amazon.com/redshift/latest/mgmt/copy-unload-iam-role.html)。 我們建議的最佳實務是，將許可政策附加到 IAM 角色，然後根據需要將其指派給使用者和群組。如需詳細資訊，請參閱 [Amazon Redshift 中的身分和存取管理](https://docs.aws.amazon.com/redshift/latest/mgmt/redshift-iam-authentication-access-control.html)。 ## 與 整合 AWS Secrets Manager 您可以從 AWS Secrets Manager中儲存的機密擷取 Redshift 的使用者名稱和密碼憑證。若要自動提供 Redshift 憑證，請使用 `secret.id` 參數。如需如何建立 Redshift 憑證機密的相關資訊，請參閱[建立 AWS Secrets Manager 資料庫機密](https://docs.aws.amazon.com/secretsmanager/latest/userguide/create_database_secret.html)。 [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/redshift/latest/mgmt/redshift-spark-connector-authentication.html) **注意** 確認：本文件包含由 [Apache Software Foundation](http://www.apache.org/) 根據 [Apache 2.0 授權](https://www.apache.org/licenses/LICENSE-2.0)所開發的範例程式碼和語言。 # 透過下推功能改善效能 Spark 連接器會自動套用述詞和查詢下推功能，以針對效能進行最佳化。這項支援表示，如果您在查詢中使用受支援的函數，Spark 連接器便會將該函數轉換為 SQL 查詢，並在 Amazon Redshift 中執行該查詢。此最佳化可減少所擷取的資料，因此 Apache Spark 可以處理較少的資料，獲得較好的效能。依預設，系統會自動啟動下推功能。若要加以停用，請將 `autopushdown` 設定為 false。 ``` import sqlContext.implicits._val sample= sqlContext.read .format("io.github.spark_redshift_community.spark.redshift") .option("url",jdbcURL ) .option("tempdir", tempS3Dir) .option("dbtable", "event") .option("autopushdown", "false") .load() ``` 下推功能支援下列函數。如果您使用的函數不在此清單中，Spark 連接器會在 Spark 中執行此函數，而不是在 Amazon Redshift 中執行，因此無法獲得最佳效能。如需 Spark 中函數的完整清單，請參閱[內建函數](https://spark.apache.org/docs/latest/api/sql/index.html)。 + 彙總函數 + avg + count + max + min + sum + stddev\$1samp + stddev\$1pop + var\$1samp + var\$1pop + 布林值運算子 + in + isnull + isnotnull + contains + endswith + startswith + 邏輯運算子 + 及 + 或 + not (or \$1) + 數學函式 + \$1 + - + \$1 + / + - (unary) + abs + acos + asin + atan + ceil + cos + exp + floor + greatest + least + log10 + pi + pow + round + sin + sqrt + tan + 其他函數 + cast + coalesce + decimal + 如 + in + 關係運算子 + \$1= + = + > + >= + < + <= + 字串函數 + ascii + lpad + rpad + translate + upper + lower + length + trim + ltrim + rtrim + like + substring + concat + 日期和時間函數 + add\$1months + date + date\$1add + date\$1sub + date\$1trunc + timestamp + trunc + 數學運算 + CheckOverflow + PromotePrecision + 關係運算 + Aliases (例如 AS) + CaseWhen + Distinct + InSet + Joins 和 cross joins + 限制 + Unions, union all + ScalarSubquery + Sorts (遞增和遞減) + UnscaledValue # 其他組態選項 在此頁面上，您可以找到可為 Amazon Redshift Spark 連接器指定之選項的說明。 ## 字串欄的大小上限 在建立儲存為 VARCHAR(256) 的資料表時，Redshift 會將字串資料欄建立為文字資料欄。如果你想要支援更大尺寸的資料欄，則可以使用 maxlength 來指定字串資料欄的最大長度。以下範例說明如何指定 `maxlength`。 ``` columnLengthMap.foreach { case (colName, length) => val metadata = new MetadataBuilder().putLong("maxlength", length).build() df = df.withColumn(colName, df(colName).as(colName, metadata)) } ``` ## 資料欄類型 若要設定資料欄類型，請使用 `redshift_type` 欄位。 ``` columnTypeMap.foreach { case (colName, colType) => val metadata = new MetadataBuilder().putString("redshift_type", colType).build() df = df.withColumn(colName, df(colName).as(colName, metadata)) } ``` ## 欄的壓縮編碼 若要在資料欄上使用特定的壓縮編碼，請使用編碼欄位。如需支援壓縮編碼的完整清單，請參閱[壓縮編碼](https://docs.aws.amazon.com/redshift/latest/dg/c_Compression_encodings.html)。 ## 欄的說明 若要設定描述，請使用 `description` 欄位。 ## Redshift 和 Amazon S3 之間的身分驗證 根據預設，結果會以 parquet 格式卸載到 Amazon S3。若要將結果卸載為以豎線分隔的文字檔案，請指定下列選項。 ``` .option("unload_s3_format", "TEXT") ``` ## Pushdown 陳述式 [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/redshift/latest/mgmt/spark-redshift-connector-other-config.html) ## 連接器參數 Spark SQL 中的參數映射或 `OPTIONS` 支援以下設定。 [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/redshift/latest/mgmt/spark-redshift-connector-other-config.html) **注意** 確認：本文件包含由 [Apache Software Foundation](http://www.apache.org/) 根據 [Apache 2.0 授權](https://www.apache.org/licenses/LICENSE-2.0)所開發的範例程式碼和語言。 # 支援的資料類型 Spark 連接器支援 Amazon Redshift 中的下列資料類型。如需 Amazon Redshift 中支援的資料類型完整清單，請參閱[資料類型](https://docs.aws.amazon.com//redshift/latest/dg/c_Supported_data_types.html)。如果資料類型不在下表中，則表示 Spark 連接器不支援該資料類型。 [\[See the AWS documentation website for more details\]](http://docs.aws.amazon.com/zh_tw/redshift/latest/mgmt/spark-redshift-connector-data-types.html) ## 複雜資料類型 您可以使用 Spark 連接器在 Redshift SUPER 資料類型資料欄中讀取和寫入 Spark 複雜資料類型 (例如 `ArrayType`、`MapType` 和 `StructType`)。如果您在讀取操作期間提供結構描述，資料欄中的資料會轉換為其在 Spark 中對應的複雜類型，包括任何巢狀類型。此外，如果已啟用 `autopushdown`，則巢狀屬性、映射值和陣列索引的投影會向下推送至 Redshift，在只存取一部分資料時，就不再需要卸載整個巢狀資料結構。 當您從連接器寫入 DataFrames 時，任何類型為 `MapType` (使用`StringType`)、`StructType`、或 `ArrayType` 的資料欄都會寫入到 Redshift SUPER 資料類型資料欄。在撰寫這些巢狀資料結構時，`tempformat` 參數的類型必須是 `CSV`、`CSV GZIP` 或 `PARQUET`。使用 `AVRO` 會導致例外狀況。撰寫具有 `StringType` 以外索引鍵類型的 `MapType` 資料結構也會導致例外狀況。 ### StructType 下列範例會示範如何建立具有 SUPER 資料類型且包含以下結構的資料表 ``` create table contains_super (a super); ``` 然後，您可以使用連接器，在使用如下列範例中的結構描述的資料表中，從 SUPER 資料欄 `a` 查詢 `StringType` 欄位 `hello`。 ``` import org.apache.spark.sql.types._ val sc = // existing SparkContext val sqlContext = new SQLContext(sc) val schema = StructType(StructField("a", StructType(StructField("hello", StringType) ::Nil)) :: Nil) val helloDF = sqlContext.read .format("io.github.spark_redshift_community.spark.redshift") .option("url", jdbcURL ) .option("tempdir", tempS3Dir) .option("dbtable", "contains_super") .schema(schema) .load().selectExpr("a.hello") ``` 下列範例會示範如何將結構寫入到資料欄 `a`。 ``` import org.apache.spark.sql.types._ import org.apache.spark.sql._ val sc = // existing SparkContext val sqlContext = new SQLContext(sc) val schema = StructType(StructField("a", StructType(StructField("hello", StringType) ::Nil)) :: Nil) val data = sc.parallelize(Seq(Row(Row("world")))) val mydf = sqlContext.createDataFrame(data, schema) mydf.write.format("io.github.spark_redshift_community.spark.redshift"). option("url", jdbcUrl). option("dbtable", tableName). option("tempdir", tempS3Dir). option("tempformat", "CSV"). mode(SaveMode.Append).save ``` ### MapType 如果您偏好使用 `MapType` 來表示資料，則可以在結構描述中使用 `MapType` 資料結構，並擷取映射中與索引鍵對應的值。請注意，`MapType` 資料結構中的所有索引鍵都必須是字串類型，並且所有值都必須是相同類型，例如 int。 下列範例會示範如何取得資料欄 `a` 中索引鍵 `hello` 的值。 ``` import org.apache.spark.sql.types._ val sc = // existing SparkContext val sqlContext = new SQLContext(sc) val schema = StructType(StructField("a", MapType(StringType, IntegerType))::Nil) val helloDF = sqlContext.read .format("io.github.spark_redshift_community.spark.redshift") .option("url", jdbcURL ) .option("tempdir", tempS3Dir) .option("dbtable", "contains_super") .schema(schema) .load().selectExpr("a['hello']") ``` ### ArrayType 如果資料欄包含陣列而不是結構，您可以使用連接器來查詢陣列中的第一個元素。 ``` import org.apache.spark.sql.types._ val sc = // existing SparkContext val sqlContext = new SQLContext(sc) val schema = StructType(StructField("a", ArrayType(IntegerType)):: Nil) val helloDF = sqlContext.read .format("io.github.spark_redshift_community.spark.redshift") .option("url", jdbcURL ) .option("tempdir", tempS3Dir) .option("dbtable", "contains_super") .schema(schema) .load().selectExpr("a[0]") ``` ### 限制 搭配 Spark 連接器使用複雜資料類型時有下列限制： + 所有巢狀的結構欄位名稱和映射索引鍵都必須小寫。如果要查詢包含大寫字母的複雜欄位名稱，因應措施是您可以嘗試省略結構描述，並使用 `from_json` Spark 函數在本機轉換傳回的字串。 + 讀取或寫入操作中使用的任何映射欄位都必須只有 `StringType` 索引鍵。 + 只有 `CSV`、`CSV GZIP` 和 `PARQUET ` 才是支援將複雜類型寫入到 Redshift 的 tempformat 值。嘗試使用 `AVRO ` 會擲回例外狀況。