3 serverless数据分析

大纲

  • 3 serverless数据分析
    • 3.1 创建Lambda
    • 3.2 创建API Gateway
    • 3.3 结果
    • 3.4 总结

3.1 创建Lambda

在Lambda中，我们将使用python3作为代码语言。

步骤图例
1、入口
2、创建（我们选择使用python3.7）
3、IAM权限（权限可信实体需要包括Lambda才能将角色绑定到Lambda上）见下方“IAM可信实体描述”
4、指定处理函数（处理程序要为用户程序的入口）
5、添加层（层为我们的代码运行时的环境，并且，兼容运行时要包含上一步中的运行时环境）
6、代码（在此代码中使用了boto3来连接Athena，可自定义sql，使用方法请看官方文档）见下方“Lambda代码”

IAM可信实体描述：

{"Version": "2012-10-17","Statement": [{"Effect": "Allow","Principal": {"Service": "lambda.amazonaws.com"},"Action": "sts:AssumeRole"}]}

Lambda代码

import boto3, os, jsonimport pandas as pdfrom pyathena import connectimport timeREGION = "us-west-2"# expected request: anomaly/{meter_id}" />def lambda_handler(event, context):ATHENA_OUTPUT_BUCKET = "【待替换S3桶路径】/athena"DB_SCHEMA = "suzikuo_test_db"USE_WEATHER_DATA = 0pathParameter = event["pathParameters"]queryParameter = event["queryStringParameters"]METER_ID = pathParameter['meter_id']DATA_START = queryParameter['data_start']DATA_END = queryParameter['data_end']OUTLIER_ONLY = queryParameter['outlier_only']query = '''select * from "{}".reading_type_intwhere meter_id = '{}'and cast(reading_date_time as timestamp) >= timestamp '{}' and cast(reading_date_time as timestamp) < timestamp '{}''''.format(DB_SCHEMA, METER_ID, DATA_START, DATA_END)athena = boto3.client('athena')response = athena.start_query_execution(QueryString=query,QueryExecutionContext={'Database': 'suzikuo_test_db'},ResultConfiguration={'OutputLocation': 's3://suzikuo-test-2022-8-4-s3/athena','EncryptionConfiguration': {'EncryptionOption': 'SSE_S3'}})while True:try:query_results = athena.get_query_results(QueryExecutionId=response['QueryExecutionId'])breakexcept Exception as err:if 'Query has not yet finished' in str(err):time.sleep(3)else:raise(err)return query_results['ResultSet']['Rows']

3.2 创建API Gateway

使用API Gateway+Lambda 可轻松实现一个serverless架构。

步骤图例
1、入口
2、API（我们使用的是Lambda，所以选HTTP API）
3、创建集成（指定要绑定的Lambda）
4、配置路由（指定路由要请求的集成（lambda））
5、一直下一步即可

3.3 结果

此案例只查询了某一ID的某个时间段内的数据
通过获取URI和参数，在Lambda中编写逻辑，可以实现我们对数据的任意操作。

3.4 总结

到此，我们已经完成了基于Glue ETL(提取、转换和加载)的serviceless 数据分析的全部过程了。在此案例中，我们使用到了AWS 服务中的Glue、S3、APIGateway、Lambda等服务实现了一个通过API访问的数据统计与分析接口。