减少数据扫描量

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

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

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

要减少为每个小文件创建 Apache Spark 任务时产生的过多并行度，请使用文件分组。 DynamicFrames这种方法减少了 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 读取器可以通过提供仅检索特定块的Range选项GetObjectAPI来检索 S3 对象的拆分。请考虑下图，看看这在实践中是如何运作的。

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

文件格式-使用分栏格式。Apache Parq uet 和 Apache ORC 是流行的列式数据格式。通过采用基于列的压缩、根据每列的数据类型对每列进行编码和压缩，从而有效地ORC存储和存储数据。有关 Parquet 编码的更多信息，请参阅 Par quet 编码定义。Parquet 文件也可以拆分。

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

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

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

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

对于亚马逊 Redshift，请使用 AWS Glue 4.0 或更高版本来利用亚马逊 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"}
)

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

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

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