减少数据扫描量

文件大小：我们建议将输入和输出的文件大小保持在适中的范围内（例如 128 MB）。文件过小和文件过大都可能会导致问题。

大量小文件会导致以下问题：

另一方面，如果您的文件类型不可分割（例如 gzip），并且文件太大，则 Spark 应用程序必须等待，直到单个任务完成读取整个文件。

要减少为每个小文件创建 Apache Spark 任务时产生的过多并行度，请使用对 DynamicFrame 文件分组。此方法降低了 Spark 驱动程序出现 OOM 异常的几率。要配置文件分组，请设置 groupFiles 和 groupSize 参数。以下代码示例在 ETL 脚本中使用 AWS Glue DynamicFrame API 及这些参数。

dyf = glueContext.create_dynamic_frame_from_options("s3",
    {'paths': ["s3://input-s3-path/"],
    'recurse':True,
    'groupFiles': 'inPartition',
    'groupSize': '1048576'},
    format="json")

压缩：如果您的 S3 对象大小达到数百兆字节，请考虑对其进行压缩。有多种压缩格式，大致可以分为两类：

对于 Spark（以及其他常见的分布式处理引擎），您需要将源数据文件分割成引擎可以并行处理的块。这些单元通常称为分割单元。在您的数据采用可分割格式后，优化的 AWS Glue 读取器可通过向 GetObject API 提供仅检索特定数据块的 Range 选项来检索 S3 对象的分割单元。请参考下图以了解实际工作原理。

只要文件为最佳大小或者文件可分割，压缩数据便可显著加快应用程序的速度。较小的数据大小会减少从 Amazon S3 扫描的数据量以及从 Amazon S3 到 Spark 集群的网络流量。另一方面，压缩和解压缩数据需要更多的 CPU。所需的计算量与压缩算法的压缩率成正比。在选择可分割的压缩格式时，请考虑这种权衡取舍。

文件格式：使用列式格式。Apache Parquet 和 Apache ORC 是常见的列式数据格式。Parquet 和 ORC 通过采用基于列的压缩、根据每列的数据类型对其进行编码和压缩，从而高效地存储数据。有关 Parquet 编码的更多信息，请参阅 Parquet encoding definitions。Parquet 文件也可分割。

列式格式按列对值进行分组，并将其一起存储在数据块中。使用列式格式时，您可以跳过对应不打算使用的列的数据块。Spark 应用程序只能检索您需要的列。通常，更好的压缩率或跳过数据块意味着从 Amazon S3 读取更少的字节，从而获得更好的性能。这两种格式还支持以下下推方法来减少 I/O：

文件布局：通过根据数据的使用方式将 S3 数据存储到不同路径的对象中，您可以高效地检索相关数据。有关更多信息，请参阅 Amazon S3 文档中的使用前缀组织对象。AWS Glue 支持以 key=value 格式将密钥和值存储到 Amazon S3 前缀中，按照 Amazon S3 路径对数据进行分区。通过分区数据，您可以限制每个下游分析应用程序扫描的数据量，从而提升性能并降低成本。有关更多信息，请参阅管理 AWS Glue 中用于 ETL 输出的分区。

分区将表分成不同的部分，并根据年、月和日等列值将相关数据保存在分组文件中，如以下示例所示。

# Partitioning by /YYYY/MM/DD
s3://<YourBucket>/year=2023/month=03/day=31/0000.gz
s3://<YourBucket>/year=2023/month=03/day=01/0000.gz
s3://<YourBucket>/year=2023/month=03/day=02/0000.gz
s3://<YourBucket>/year=2023/month=03/day=03/0000.gz
...

您可以使用 AWS Glue Data Catalog 中的表进行建模，定义数据集的分区。然后，您可以使用分区修剪来限制数据扫描量，如下所示：

Amazon S3 分区过多：避免在包含大量值的列上对 Amazon S3 数据进行分区，例如包含数千个值的 ID 列。这可能会大大增加存储桶中的分区数量，因为可能的分区数量是您分区所依据的所有字段的乘积。分区过多可能会导致以下情况：

例如，当您在 partitionBy 或 partitionKeys 中设置日期类型时，yyyy/mm/dd 等日期级分区适用于许多使用案例。但是，yyyy/mm/dd/<ID> 可能会生成太多的分区，以至于对整体性能产生负面影响。

另一方面，某些使用案例（例如实时处理应用程序）需要 yyyy/mm/dd/hh 等许多分区。如果您的使用案例需要大量分区，请考虑使用 AWS Glue 分区索引来降低从数据目录检索分区元数据的延迟。

对于 DynamicFrame，请使用 sampleQuery 选项。使用 create_dynamic_frame.from_catalog 时，按如下方式配置 additional_options 参数。

query = "SELECT * FROM <TableName> where id = 'XX' AND"
datasource0 = glueContext.create_dynamic_frame.from_catalog(
  database = db,
  table_name = table,
  additional_options={
      "sampleQuery": query,
      "hashexpression": key,
      "hashpartitions": 10,
      "enablePartitioningForSampleQuery": True
    },
  transformation_ctx = "datasource0"
)

using create_dynamic_frame.from_options 时，按如下方式配置 connection_options 参数。

query = "SELECT * FROM <TableName> where id = 'XX' AND"
datasource0 = glueContext.create_dynamic_frame.from_options(
    connection_type = connection, 
    connection_options={
        "url": url,
        "user": user,
        "password": password,
        "dbtable": table,
        "sampleQuery": query,
        "hashexpression": key,
        "hashpartitions": 10,
        "enablePartitioningForSampleQuery": True
    }
)

对于 DataFrame，使用 query 选项。

query = "SELECT * FROM <TableName> where id = 'XX'"
jdbcDF = spark.read \
.format('jdbc') \
.option('url', url) \
.option('user', user) \
.option('password', pwd) \
.option('query', query) \
.load()

对于 Amazon Redshift，使用 AWS Glue 4.0 或更高版本，以利用 Amazon Redshift Spark 连接器中的下推支持。

dyf = glueContext.create_dynamic_frame.from_catalog(
    database = "redshift-dc-database-name", 
    table_name = "redshift-table-name", 
    redshift_tmp_dir = args["temp-s3-dir"], 
    additional_options = {"aws_iam_role": "arn:aws:iam::role-account-id:role/rs-role-name"}
)

对于其他数据库，请查阅该数据库的文档。

要避免对所有连续的作业运行进行全面扫描，并仅处理上次作业运行期间不存在的数据，请启用作业书签。

要限制要处理的输入数据的数量，请使用作业书签启用有界执行。这有助于减少每次作业运行扫描的数据量。