使用AWS Glue与AWS Kinesis构建的流式ETL作业（一）——数据实时采集

本文介绍: 在此案例中，我们使用了Clo u dWatc h + Kin e si s Data St re a m 完成了前期的数据实时采集的工作，并且，使用了La mb da来作为触发器来对数据进行了一个验证操作（也可使用La mb da来进行ETL工作）。

更高效的从项目的数据集中提取有意义的数据，并进行统计分析。

AWS Glue中的流式ETL是基于Apac he Spark的结构化流引擎。该引擎提供一种高容错、可扩展且易于实现的方法，能够实现端到端的流处理。

在此流式ETL架构中，将使用AWS Lam bda模拟创建日志和创建AWS Clo u dWatc h 指标，并将其以流的形式发布至AWS Kin e si s Data St reams中。我们还将在AWS Glue中创建一项流式ETL作业，该作业以微批次（间隔性批次处理）的形式获取连续生成的stream 数据，并对数据进行转换、聚合，最后将结果传递至接收器。开发人员利用这部分结果生成可视化图表或在下游流程中继续使用。

在这里插入图片描述

我们使用AWS Kinesi s Data St ream来实时捕获数据，它可以从数十万个数据源提取并存储数据流，其中包括：

步骤	图例
1、入口
2、创建（按需模式无需手动预置和扩展数据流）

步骤	图例
1、入口
2、选择上步中创建的AWS Kinesi s
3、AWS IAM角色（需要有AWS Kin es i s Data Stream的权限）	权限与实体见下方“AWS IAM角色权限”
4、配置筛选条件（可根据日志格式自定义）（例如：图中配置为筛选包含“is _save _kinesis“的数据）
5、测试数据（可以选定某条日志流，或自定义数据进行测试结果显示）
6、完成日志筛选条件创建（每个日志组最多只能创建两条）

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Principal": {
                "Service": "logs.【区域】.amazonaws.com"
            },
            "Action": "sts:AssumeRole",
            "Condition": {
                "StringLike": {
                    "aws:SourceArn": "【CloudWatch的ARN】"
                }
            }
        }
     ]
}

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "WriteOutputKinesis",
            "Effect": "Allow",
            "Action": [
                "kinesis:DescribeStream",
                "kinesis:PutRecord",
                "kinesis:PutRecords"
            ],
            "Resource": [
                "【Kinesis Data Stream的ARN】"
            ]
        }
    ]
}

def lambda_handler(event, context):
    raw_kinesis_records = event['Records']
    # records = deaggregate_records(raw_kinesis_records)
    records = raw_kinesis_records
    for record in records:
        #Kinesis data is base64 encoded so decode here
        payload = base64.b64decode(record["kinesis"]["data"], validate = False)
        data = gzip.decompress(payload).decode("utf-8")
		print(data)