aws(学习笔记第三十三课) 深入使用cdk

• 使用cdk生成athena以及aws glue crawler

学习内容：

• 使用aws athena + aws glue crawler

---

1. 使用aws athena

1.1 什么是aws athena

aws athena是aws提供的数据分析service，可以使用SQL语言对S3上保存的数据进行分析。

• managed service，所以不需要维护。
• 基于OpenSource的框架构筑
• 基于处理的数据量进行收费
• 对数据提供加密功能
  注意 和RDB不能进行JOIN操作，所以只能提供对csv和json进行数据查询

1.2 什么是aws glue

aws glue是aws提供的managed ETL service。能够简单的进行分析数据的准备和load。table和schema关联的metadata能够作为aws glue catalog data进行保存。

1.2 为什么aws athena和aws glue一起使用

aws athena结合aws glue能够将aws glue作成的database或者schema，使用aws athena进行查询。

2. 开始练习aws athena

2.1 代码链接

代码链接aws-cdk-examples

2.2 整体架构

2.3 代码解析

2.3.1 创建测试数据的S3 bucket

 # creating the buckets where the logs will be placed
 logs_bucket = s3.Bucket(self, 'logs-bucket',
             bucket_name=f"auditing-logs-{self.account}",
             removal_policy=RemovalPolicy.DESTROY,
             auto_delete_objects=True
 )

2.3.2 创建保存查询结果的S3 bucket

 # creating the bucket where the  queries output will be placed
 query_output_bucket = s3.Bucket(self, 'query-output-bucket',
      bucket_name=f"auditing-analysis-output-{self.account}",
      removal_policy=RemovalPolicy.DESTROY,
       auto_delete_objects=True
    )

2.3.3 将示例的程序json数据文件同期到S3 bucket

# uploading the log files to the bucket as examples
  s3_deployment.BucketDeployment(self, 'sample-files',
         destination_bucket=logs_bucket,
         sources=[s3_deployment.Source.asset('./log-samples')],
         content_type='application/json',
         retain_on_delete=False
)

2.3.4 创建aws glue的cfnDatabase

    # creating the Glue Database to serve as our Data Catalog
    glue_database = glue.CfnDatabase(self, 'log-database',
            catalog_id=self.account,
            database_input=glue.CfnDatabase.DatabaseInputProperty(
            name="log-database"))

2.3.5 创建aws glue crawler需要的权限Role

# creating the permissions for the crawler to enrich our Data Catalog
        glue_crawler_role = iam.Role(self, 'glue-crawler-role',
                                     role_name='glue-crawler-role',
                                     assumed_by=iam.ServicePrincipal(service='glue.amazonaws.com'),
                                     managed_policies=[
                                         # Remember to apply the Least Privilege Principle and provide only the permissions needed to the crawler
                                         iam.ManagedPolicy.from_managed_policy_arn(self, 'AmazonS3FullAccess',
                                                                                   'arn:aws:iam::aws:policy/AmazonS3FullAccess'),
                                         iam.ManagedPolicy.from_managed_policy_arn(self, 'AWSGlueServiceRole',
                                                                                   'arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole')
                                     ])

这里需要两个policy，AmazonS3FullAccess和AWSGlueServiceRole。

2.3.6 创建aws glue crawler

 # creating the Glue Crawler that will automatically populate our Data Catalog. Don't forget to run the crawler
        # as soon as the deployment finishes, otherwise our Data Catalog will be empty. Check out the README for more instructions
        glue.CfnCrawler(self, 'logs-crawler',
                        name='logs-crawler',
                        database_name=glue_database.database_input.name,
                        role=glue_crawler_role.role_name,
                        targets={
                            "s3Targets": [
                                {"path": f's3://{logs_bucket.bucket_name}/products'},
                                {"path": f's3://{logs_bucket.bucket_name}/users'}
                            ]
                        })

这里，aws glue crawler执行ETL Extract Transform Load，将S3 bucket里面的products和users的数据文件，经过转换将json数据文件load到glue database。

2.3.7 创建aws athena work group

# creating the Athena Workgroup to store our queries
        work_group = athena.CfnWorkGroup(self, 'log-auditing-work-group',
                                         name='log-auditing',
                                         work_group_configuration=athena.CfnWorkGroup.WorkGroupConfigurationProperty(
                                             result_configuration=athena.CfnWorkGroup.ResultConfigurationProperty(
                                                 output_location=f"s3://{query_output_bucket.bucket_name}",
                                                 encryption_configuration=athena.CfnWorkGroup.EncryptionConfigurationProperty(
                                                     encryption_option="SSE_S3"
                                                 ))))

aws athena通过work group进行管理，创建了workgroup之后，在里面继续创建query。

2.3.8 创建aws athena query

# creating an example query to fetch all product events by date
        product_events_by_date_query = athena.CfnNamedQuery(self, 'product-events-by-date-query',
                                                            database=glue_database.database_input.name,
                                                            work_group=work_group.name,
                                                            name="product-events-by-date",
                                                            query_string="SELECT * FROM \"log-database\".\"products\" WHERE \"date\" = '2024-01-19'")

        # creating an example query to fetch all user events by date
        user_events_by_date_query = athena.CfnNamedQuery(self, 'user-events-by-date-query',
                                                         database=glue_database.database_input.name,
                                                         work_group=work_group.name,
                                                         name="user-events-by-date",
                                                         query_string="SELECT * FROM \"log-database\".\"users\" WHERE \"date\" = '2024-01-22'")

        # creating an example query to fetch all events by the user ID
        all_events_by_userid_query = athena.CfnNamedQuery(self, 'all-events-by-userId-query',
                                                          database=glue_database.database_input.name,
                                                          work_group=work_group.name,
                                                          name="all-events-by-userId",
                                                          query_string="SELECT * FROM (\n"
                                                                       "    SELECT transactionid, userid, username, domain, datetime, action FROM \"log-database\".\"products\" \n"
                                                                       "UNION \n"
                                                                       "    SELECT transactionid, userid, username, domain, datetime, action FROM \"log-database\".\"users\" \n"
                                                                       ") WHERE \"userid\" = '123'")

2.3.9 调整执行顺序

# adjusting the resource creation order
product_events_by_date_query.add_dependency(work_group)
user_events_by_date_query.add_dependency(work_group)
all_events_by_userid_query.add_dependency(work_group)

2.4 开始执行aws cdk for athena

2.4.1 执行部署

python -m venv .venv
source .venv/Scripts/activate # windows platform
pip install -r requirements.txt
cdk synth
cdk --require-approval never deploy

2.4.2 执行crawler爬虫

默认crawler是不启动的，需要run起来。

正常执行完毕。数据都由S3 bucket的json文件，经过ETL，进入到aws glue database里面了。

2.4.3 查看aws athena的queries

AWS Athena > 查询编辑器 > 已保存的查询 > 工作组 > log auditing

2.4.4 执行aws athena的queries

2.4.5 查看aws athena的queries执行结果