使用 AWS Glue ETL 中的下推來最佳化讀取

下推是一種最佳化技術，可將有關擷取資料的邏輯推送到更接近資料來源的位置。來源可以是 Amazon S3 之類的資料庫或檔案系統。直接在來源上執行某些操作時，您可以不透過網路，將所有資料傳送至 AWS Glue 管理的 Spark 引擎，以節省時間和提高處理能力。

這個的另一種說法是下推減少了資料掃描。如需有關識別此技術何時適用的程序的詳細資訊，請參閱《AWS 規定指南》中 AWS Glue for Apache Spark 任務效能調校最佳實務指引中的減少資料掃描數量。

對 Amazon S3 上存放的檔案進行述詞下推

在 Amazon S3 上使用依字首整理的檔案時，您可以透過定義下推述詞，篩選目標 Amazon S3 路徑。您可以直接將篩選條件套用至 AWS Glue Data Catalog 中存放的分割區中繼資料，而不是在 DynamicFrame 中讀取完整的資料集並套用篩選條件。這種方法可讓您選擇性地僅列出和讀取必要的資料。如需有關此程序的詳細資訊，包括依分割區寫入儲存貯體，請參閱 在 AWS Glue 中管理適用於 ETL 輸出的分割區。

您可以使用 push_down_predicate 參數在 Amazon S3 中達成述詞下推。考慮您在 Amazon S3 中依年份、月份和日期分割的儲存貯體。如果您想擷取 2022 年 6 月的客戶資料，可以指示 AWS Glue 僅讀取相關的 Amazon S3 路徑。在本案例中 push_down_predicate 為 year='2022' and month='06'。綜合來說，您可以依如下方式實現讀取操作：

Python

customer_records = glueContext.create_dynamic_frame.from_catalog( 
    database = "customer_db", 
    table_name = "customer_tbl",
    push_down_predicate = "year='2022' and month='06'"
)

Scala

val customer_records = glueContext.getCatalogSource(
database="customer_db", 
tableName="customer_tbl", 
pushDownPredicate="year='2022' and month='06'"
).getDynamicFrame()

在先前的案例中，push_down_predicate 會從 AWS Glue Data Catalog 擷取所有分割區的清單，並在讀取基礎 Amazon S3 檔案之前進行篩選。雖然這在大多數情況下很有用，但在處理具有數百萬個分割區的資料集時，分割區的列出程序可能很耗時。為了解決這個問題，可以使用伺服器端的分割區剔除來改善效能。此操作透過在 AWS Glue Data Catalog 中為資料建立分割區索引來完成。如需有關分割區索引的詳細資訊，請參閱 建立分割區索引 。然後，您可以使用 catalogPartitionPredicate 選項參考索引。如需使用 catalogPartitionPredicate 擷取分割區的範例，請參閱 使用目錄分割述詞的伺服器端篩選。

使用 JDBC 來源時下推

GlueContext 中使用的 AWS Glue JDBC 讀取器透過提供可直接在來源上執行的自訂 SQL 查詢，協助在支援的資料庫上進行下推。可以透過設定 sampleQuery 參數來實現這一點。範例查詢可以指定要選取的資料欄，以及提供下推述詞來限制向 Spark 引擎傳輸的資料。

範例查詢依預設會在單一節點上運作，但處理大量資料時可能會導致任務失敗。若要使用此功能大規模查詢資料，您應該將 enablePartitioningForSampleQuery 設定為 true 來設定查詢分割區，如此會將查詢分散到所選索引鍵之間的多個節點。查詢分割區還需要一些其他必要的組態參數。如需有關查詢分割區的詳細資訊，請參閱 從 JDBC 資料表中平行讀取。

設定 enablePartitioningForSampleQuery 時，AWS Glue 會在查詢資料庫時將下推述詞與分割述詞相結合。sampleQuery 的結尾必須是 AND，以便 AWS Glue 附加分割條件。(如果您沒有提供下推述詞，則 sampleQuery 的結尾必須是 WHERE)。請參閱下面的範例，其中我們向下推送一個述詞，以僅擷取 id 大於 1000 的資料列。此 sampleQuery 僅會傳回其中 id 大於指定值的資料列名稱和位置資料欄：

Python

sample_query = "select name, location from customer_tbl WHERE id>=1000 AND"
customer_records = glueContext.create_dynamic_frame.from_catalog(
    database="customer_db",
    table_name="customer_tbl",
    sample_query = "select name, location from customer_tbl WHERE id>=1000 AND",

    additional_options = { 
                           "hashpartitions": 36 , 
                           "hashfield":"id",
                           "enablePartitioningForSampleQuery":True, 
                           "sampleQuery":sample_query
                          }
)

Scala

val additionalOptions = Map( 
        "hashpartitions" -> "36", 
        "hashfield" -> "id", 
        "enablePartitioningForSampleQuery" -> "true", 
        "sampleQuery" -> "select name, location from customer_tbl WHERE id >= 1000 AND"
        )
 
    val customer_records = glueContext.getCatalogSource(
        database="customer_db", 
        tableName="customer_tbl").getDynamicFrame()
                    

注意

如果customer_tbl資料目錄和基礎資料存放區中的名稱不同，您必須在 sample_query 中提供基礎資料表名稱，因為查詢會傳遞至基礎資料存放區。

您也可以根據 JDBC 資料表進行查詢，而不需與 AWS Glue Data Catalog 整合。您可以透過提供 useConnectionProperties 和 connectionName 來重複使用來自預先存在之連線的憑證，而不是將使用者名稱和密碼作為參數提供給該方法。在本範例中，我們從名為 my_postgre_connection 的連線擷取憑證。

Python

connection_options_dict = {
    "useConnectionProperties": True,
    "connectionName": "my_postgre_connection",
    "dbtable":"customer_tbl",
    "sampleQuery":"select name, location from customer_tbl WHERE id>=1000 AND",
    "enablePartitioningForSampleQuery":True,
    "hashfield":"id",
    "hashpartitions":36
    }

customer_records = glueContext.create_dynamic_frame.from_options(
    connection_type="postgresql",
    connection_options=connection_options_dict
    )

Scala

val connectionOptionsJson = """
      {
        "useConnectionProperties": true,
        "connectionName": "my_postgre_connection",
        "dbtable": "customer_tbl",
        "sampleQuery": "select name, location from customer_tbl WHERE id>=1000 AND",
        "enablePartitioningForSampleQuery" : true,
        "hashfield" : "id",
        "hashpartitions" : 36
      }
    """
    
    val connectionOptions = new JsonOptions(connectionOptionsJson)
    
    val dyf = glueContext.getSource("postgresql", connectionOptions).getDynamicFrame()

AWS Glue 中下推的注意事項和限制

就概念來說，下推在從非串流來源讀取時適用。AWSGlue 支援多種來源，下推的能力取決於來源和連接器。

• 連線到 Snowflake 時，您可以使用 query 選項。在 AWS Glue 4.0 及更新版本的 Redshift 連接器中存在類似功能。如需有關使用 query 從 Snowflake 讀取的詳細資訊，請參閱 從 Snowflake 資料表讀取。

• DynamoDB ETL 讀取器不支援篩選條件或下推述詞。MongoDB 和 DocumentDB 也不支援這種功能。

• 從以開放資料表格式存放在 Amazon S3 中的資料進行讀取時，Amazon S3 中檔案的分割方法已不再足夠。若要使用開放資料表格式從分割區讀取和寫入，請參閱該格式的文件。

• DynamicFrame 方法不會執行 Amazon S3 投影下推。將從通過述詞篩選條件的檔案讀取所有資料欄。

• 使用 AWS Glue 中的 custom.jdbc 連接器時，下推的能力取決於來源和連接器。請檢閱適當的連接器文件，以確認連接器是否以及如何支援 AWS Glue 中的下推。