使用 Glue 和 Athena 分析 AWS 服务日志

简介

在业务上线初期或业务的推广期，许多客户都有分析访问流量，用以区分正常流量和异常流量的的需求，从而识别当前是否业务遭遇了流量攻击。AWS的许多服务本身会产生日志，比如ALB负载均衡访问日志，CloudFront CDN服务访问日志，VPC流日志等。可以打开并抓取这些日志进行分析, 了解当前的访问情况。但这些日志的规模一般都是巨大的，采用下载后进行静态的分析方式非常困难，推荐使用数据查询服务比如AWS Athena进行SQL查询。本文以VPC流日志为例，指导如何使用Glue爬网程序构建VPC流日志的数据目录，并使用GLUE ETL 作业把源数据进行分区并转换成Parquet格式，加速Athena查询VPC流日志的速度并减少扫描量以降低分析的成本。

架构

一. 创建VPC流的数据目录

创建VPC Flow logs。可以根据需求选择打开整体VPC的日志，或指定子网的日志，或指定ENI网络接口的日志。VPC flow logs会把数据发送到您指定的S3存储桶中。我们可以让流日志抓取所有类型的流量，聚合间隔可以选择1分钟以更快速的看到数据。
添加分类器。由于VPC Flow logs的文件内容中，每列使用空格作为间隔，而Glue爬网程序自带的分类器是按照逗号分隔做文件解析，所以需要先添加Glue爬网程序自定义分类器。分类器类型使用CSV, 空格分隔符，引用符号选双引号，使用列标题检测

添加AWS Glue爬网程序，进行数据目录的创建，分类器选用之前建立的自定义分类器.

4. 继续配置爬网程序的其他项目。这里需要注意的是

数据存储填写VPC flow logs的位置，比如s3://bucketname/AWSLogs/accountId/vpcflowlogs/
赋予爬网程序的IAM Role必须具有相应的权限，让其可以访问S3中的vpc流日志数据。

VPC 流日志所在S3的访问权限

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:PutObject"
            ],
            "Resource": [
                "arn:aws:s3:::vpcflowlogtestdata/AWSLogs/accountID/vpcflowlogs/*"
            ]
        }
    ]
}

运行Glue爬网程序, 可以看到数据目录，Glue会默认把S3中的层级按照分区键处理。

二. 使用Glue ETL Job生成分区数据

VPC流日志的数据规模会很大，通过对数据进行分区，在查询的时候可以减少扫描量从而加快查询速度。同时，对最终数据采用Parquet这种列式存储格式，也是加快查询速度的重要手段。我们使用Glue Spark Job来完成这种转换，最终把数据整理成 s3://convertedLogs/year=yyyy/month=mm/day=dd/action=xxx/ 这种分区结构。

Action字段代表这条vpc流日志的动作，比如ACCEPT或REJECT。一般会根据业务的查询类型，选择合适的分区键，这对扫描效率有很大的影响

VPC 流日志数据格式与样例

<version> <account-id> <interface-id> <srcaddr> <dstaddr> <srcport> <dstport> <protocol> <packets> <bytes> <start> <end> <action> <log-status>

2 123456789010 eni-1235b8ca123456789 172.31.16.139 172.31.16.21 20641 22 6 20 4249 1418530010 1418530070 ACCEPT OK

添加Glue作业, 作业类型选择Spark
Job使用的IAM Role拥有S3源、目标、临时目录的访问权限。

开启Bookmark功能。可以避免处理以往已处理过的数据，有关Bookmark介绍，可参考 https://docs.aws.amazon.com/glue/latest/dg/monitor-continuations.html

数据源，选择之前用爬网程序创建的vpcflowlogs数据库

在数据目标中创建表，选择Parquet格式

数据映射，因为我们会用新的分区键，所以可以不转换原来的partition列

编辑产生的代码，增加创建分区键的逻辑。把源数据中start字段，使用Spark的函数转换为日期信息, 并用withColumn方法为DataFrame增加year,month,day 三个字段

from awsglue.dynamicframe import DynamicFrame
from pyspark.sql.functions import *
dropnullfields3 = DropNullFields.apply(frame = resolvechoice2, transformation_ctx = "dropnullfields3")

dfdst = dropnullfields3.toDF()
dfdst = dfdst.withColumn("year",year(to_date(from_unixtime(dfdst.start))))
dfdst = dfdst.withColumn("month",month(to_date(from_unixtime(dfdst.start))))
dfdst = dfdst.withColumn("day",dayofmonth(to_date(from_unixtime(dfdst.start))))

resultFrame = DynamicFrame.fromDF(dfdst,glueContext,"result")

加入partition分区键信息，partition使用刚刚加入的year, month ,day字段和本身就存在的action字段

datasink4 = glueContext.write_dynamic_frame.from_options(frame = resultFrame, connection_type = "s3", connection_options = {"path": "s3://convertedVpcflowLog/convertedLogs/","partitionKeys":["year","month","day","action"]}, format = "parquet", transformation_ctx = "datasink4")

运行Glue ETL Job，可以看到转换后parquet文件已生成，并且的数据在S3桶内的结构已经进行了分区

三. 新建Glue爬网程序，建立新的数据目录。数组源选择经过ETL后的S3路径，如s3://convertedVpcFlowlog/convertedLogs/

爬网程序运行后，在数据目录中可以看到数据表已经建立，并且识别出了Partition信息

四. 进入AWS Athena，可以对VPC流日志进行SQL查询。

下面这个例子，查询所有被拒绝的TCP连接请求的源IP地址。

SELECT year,month,day,
  srcaddr,
  action,
  protocol
FROM "vpcflowlogs"."convertedlogs5" 
WHERE action = 'REJECT' and protocol=6 limit 100;

五．由于VPC流日志是不断生成的，所以需要创建一个工作流，持续的进行ETL的工作。我们可以通过配置Glue触发器，每隔5分钟自动调用Glue ETL Job进行数据转换。

总结

本文使用了AWS Glue Crawler进行VPC流日志的数据表创建，在处理大量数据的情况下使用Glue ETL JOB对数据进行了分区键的创建，并且使用Glue触发器每隔5分钟进行持续的ETL任务调用，Glue bookmark功能的引入也避免了处理重复的数据。整体采用无服务器的架构，采用Athena进行数据查询。客户无需购买和管理资源就可进行云上大规模的数据存储，处理，分析。

亚马逊AWS官方博客

使用 Glue 和 Athena 分析 AWS 服务日志

本篇作者