每日ETL场景下AWS Glue Crawler无法生成新表的解决方案咨询

一、实现每日生成新表的方案 #

• 按日期划分S3存储目录
  将每日解压后的独立文件存入带日期后缀的专属文件夹，例如s3://your-bucket/daily-data/2024-05-20/、s3://your-bucket/daily-data/2024-05-21/。每个日期目录仅存放当日的新数据，避免不同日期数据混杂。

• 配置爬虫生成带日期标识的新表
  在Glue爬虫配置中，修改表命名规则，添加日期前缀/后缀，比如设置表名格式为daily_table_{date}_{filename}（{date}为当日日期，{filename}为原始文件名）。同时在爬虫的「Schema变更策略」中，选择「创建新表以适配新Schema」，确保遇到新结构文件时自动生成新表，而非更新现有表。

• 用Glue Job动态创建表
  脱离爬虫自动建表的限制，通过Glue Job代码主动检测每个文件的Schema并创建对应表。示例代码如下：

  import boto3
  from awsglue.context import GlueContext
  from pyspark.context import SparkContext
  from datetime import datetime
  
  sc = SparkContext()
  glueContext = GlueContext(sc)
  glue_client = boto3.client('glue')
  
  today = datetime.today().strftime('%Y-%m-%d')
  base_s3_path = f"s3://your-bucket/daily-data/{today}/"
  
  # 列出当日目录下所有文件
  s3_client = boto3.client('s3')
  response = s3_client.list_objects_v2(Bucket='your-bucket', Prefix=f"daily-data/{today}/")
  file_keys = [obj['Key'] for obj in response.get('Contents', []) if not obj['Key'].endswith('/')]
  
  for key in file_keys:
      file_path = f"s3://your-bucket/{key}"
      # 读取文件获取Schema
      dynamic_frame = glueContext.create_dynamic_frame.from_options(
          connection_type="s3",
          connection_options={"paths": [file_path]},
          format="csv"  # 根据实际文件格式调整为json/parquet等
      )
      schema = dynamic_frame.schema()
      # 生成带日期和文件名的表名
      table_name = f"daily_{today}_{key.split('/')[-1].split('.')[0]}"
  
      # 创建Glue Catalog表
      glue_client.create_table(
          DatabaseName="your_target_db",
          TableInput={
              "Name": table_name,
              "StorageDescriptor": {
                  "Columns": [{"Name": col.name, "Type": col.dataType} for col in schema],
                  "Location": file_path,
                  "InputFormat": "org.apache.hadoop.mapred.TextInputFormat",
                  "OutputFormat": "org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat"
              },
              "TableType": "EXTERNAL_TABLE"
          }
      )
  

二、实现仅读取新数据的增量处理方案 #

• 针对单日目录的增量爬取
  用CloudWatch Events定时触发Lambda函数，在每日数据解压完成后，动态修改Glue爬虫的爬取路径为当日的日期目录，再启动爬虫。这样爬虫只会处理当日新数据，不会触及历史数据，避免覆盖现有表。

• 固定Schema文件用分区表管理
  对于Schema稳定的文件，将其存入按日期分区的目录，例如s3://your-bucket/fixed-schema-data/date=2024-05-20/。配置爬虫识别分区键date，每次爬取新分区时，会自动将新数据添加到现有表的对应分区中，无需生成新表。

• 在Glue Job中按时间筛选新文件
  直接在Job中通过S3文件的最后修改时间筛选当日新增文件，只处理这些数据：

  import boto3
  from datetime import datetime
  from awsglue.context import GlueContext
  from pyspark.context import SparkContext
  
  sc = SparkContext()
  glueContext = GlueContext(sc)
  s3_client = boto3.client('s3')
  
  bucket = "your-bucket"
  data_prefix = "uncompressed-data/"
  today_start = datetime.today().replace(hour=0, minute=0, second=0, microsecond=0)
  
  # 筛选当日修改的文件
  response = s3_client.list_objects_v2(Bucket=bucket, Prefix=data_prefix)
  new_file_paths = []
  for obj in response.get('Contents', []):
      last_modified = obj['LastModified'].replace(tzinfo=None)
      if last_modified >= today_start:
          new_file_paths.append(f"s3://{bucket}/{obj['Key']}")
  
  # 读取并处理新数据
  if new_file_paths:
      df = glueContext.create_dynamic_frame.from_options(
          connection_type="s3",
          connection_options={"paths": new_file_paths},
          format="csv"  # 适配实际文件格式
      )
      # 后续数据处理逻辑（转换、写入等）
  

关键注意事项 #

• 绝对不要让爬虫爬取包含多日数据的根目录，否则Glue会默认将所有文件合并为单表，导致新数据无法触发新表创建或更新。
• Schema频繁变化的文件优先用「每日生成新表」方案，避免Schema冲突破坏现有表结构。
• 可通过Lambda+CloudWatch Events实现全流程自动化：数据解压完成后触发Lambda，自动配置爬虫或启动Glue Job完成当日数据处理。

内容的提问来源于stack exchange，提问作者Pavan P Agarwadekar