如何提取AWS Config规则非合规资源结果至文件并实现自动化？

1. 单次提取不合规资源（测试用） #

直接用AWS CLI调用Config服务API，筛选指定规则下的NON_COMPLIANT资源，输出为JSON或CSV文件：

输出JSON格式 #

aws configservice get-compliance-details-by-config-rule \
    --config-rule-name "Required Tags" \
    --compliance-types NON_COMPLIANT \
    --output json > non_compliant_resources.json

转成Databricks友好的CSV格式 #

借助jq工具提取关键字段（资源类型、资源ID、不合规注释）并生成CSV：

aws configservice get-compliance-details-by-config-rule \
    --config-rule-name "Required Tags" \
    --compliance-types NON_COMPLIANT \
    --output json | jq -r '.evaluationResults[] | [.evaluationResultIdentifier.evaluationResultQualifier.resourceType, .evaluationResultIdentifier.evaluationResultQualifier.resourceId, .annotation] | @csv' > non_compliant_resources.csv

2. 全自动化流程实现 #

步骤1：用AWS Lambda定期拉取并存储结果 #

创建Python Lambda函数，实现自动拉取、格式转换、上传S3的逻辑：

import boto3
import csv
from io import StringIO
import datetime

def lambda_handler(event, context):
    config_client = boto3.client('configservice')
    s3_client = boto3.client('s3')
    
    # 获取不合规资源（处理分页逻辑，适配超过100条的场景）
    all_results = []
    next_token = None
    while True:
        kwargs = {
            'ConfigRuleName': 'Required Tags',
            'ComplianceTypes': ['NON_COMPLIANT']
        }
        if next_token:
            kwargs['NextToken'] = next_token
        
        response = config_client.get_compliance_details_by_config_rule(**kwargs)
        all_results.extend(response['evaluationResults'])
        
        next_token = response.get('NextToken')
        if not next_token:
            break
    
    # 生成CSV内容
    output = StringIO()
    writer = csv.writer(output)
    writer.writerow(['ResourceType', 'ResourceId', 'Annotation', 'EvaluationTimestamp'])
    
    for result in all_results:
        qualifier = result['evaluationResultIdentifier']['evaluationResultQualifier']
        writer.writerow([
            qualifier['resourceType'],
            qualifier['resourceId'],
            result.get('annotation', ''),
            result['resultRecordedTime'].isoformat()
        ])
    
    # 上传到S3，文件名带时间戳避免覆盖
    timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
    s3_key = f'aws-config-non-compliant/{timestamp}_non_compliant_resources.csv'
    s3_client.put_object(
        Bucket='your-target-s3-bucket',
        Key=s3_key,
        Body=output.getvalue()
    )
    
    return {'statusCode': 200, 'body': f'File uploaded to S3: {s3_key}'}

给Lambda配置IAM权限：允许调用configservice:GetComplianceDetailsByConfigRule，允许写入目标S3桶。

步骤2：设置定时触发 #

通过CloudWatch Events创建定时规则，按需求频率（如每日凌晨）触发Lambda函数，实现自动拉取。

步骤3：Databricks自动导入S3文件 #

在Databricks中创建定时作业，通过Notebook读取S3文件并写入Databricks表：

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName('AWSConfigNonCompliantImporter').getOrCreate()

# 读取S3路径下的所有合规文件，自动识别表头
df = spark.read.csv(
    's3://your-target-s3-bucket/aws-config-non-compliant/',
    header=True,
    inferSchema=True
)

# 追加写入Databricks表（可根据需求调整写入模式）
df.write.mode('append').saveAsTable('default.aws_config_non_compliant_resources')

设置Databricks作业的触发频率与Lambda同步，完成自动导入。

3. 可选优化 #

• 改用Parquet格式存储：在Lambda中用pandas将数据转成Parquet后上传，更适合大数据处理场景
• 添加告警机制：通过CloudWatch告警监控Lambda运行状态、S3上传失败等异常
• 字段扩展：根据需求提取更多AWS Config返回的字段（如规则ARN、合规结果时间等）

内容的提问来源于stack exchange，提问作者jlaurenza