无服务器数据工作流

使用 AWS Kinesis 与 Lambda 处理流数据

Maksim Pecherskiy

Data Engineer

回顾

回顾

使用 AWS Kinesis 与 Lambda 处理流数据

当前分区方式

当前分区

使用 AWS Kinesis 与 Lambda 处理流数据

筛除超速者

筛除超速者

使用 AWS Kinesis 与 Lambda 处理流数据

按天聚合

按天汇总

使用 AWS Kinesis 与 Lambda 处理流数据

打开 recordReaderS3

打开 recordReaderS3

使用 AWS Kinesis 与 Lambda 处理流数据

编辑 recordReaderS3

...
import pytz
from datetime import datetime
tz = pytz.timezone('America/Los_Angeles')
使用 AWS Kinesis 与 Lambda 处理流数据

编辑 recordReaderS3

def record_created_handler(event, context):

    ...
    ## 获取超过45的最高车速
    too_fast = top_speeds.loc[top_speeds.speed > SPEED_ALERT_THRESHOLD :]

    ## 生成对象键
    fdate = datetime.now(tz).strftime("%Y%m%d/%HH%MM%SS")
    obj_key = f"speeders/{fdate}.csv" # speeders 文件夹中的文件名


    ## 写入对象到 S3
    s3.put_object(Bucket='sd-vehicle-data', 
                  Key=obj_key,
                  Body=too_fast.to_csv(sep=" ", index=False)
    )
使用 AWS Kinesis 与 Lambda 处理流数据

按日期时间写入超速者

按日期和时间写入超速记录

使用 AWS Kinesis 与 Lambda 处理流数据

speederAggregator

speederAggregator

使用 AWS Kinesis 与 Lambda 处理流数据

创建 speederAggregator

创建 speederAggregator

使用 AWS Kinesis 与 Lambda 处理流数据

添加 AWS data wrangler 层

添加 AWS data wrangler 层

使用 AWS Kinesis 与 Lambda 处理流数据

speederAggregator 资源

speederAggregator 资源

使用 AWS Kinesis 与 Lambda 处理流数据

创建定时触发器

创建定时触发器

使用 AWS Kinesis 与 Lambda 处理流数据

Cron 表达式

cron(50 6 * * ? *)
  • 第 50 分钟
  • UTC 第 6 小时
  • 每月的每天
  • 每个月
  • 每周的每天
  • 每年
使用 AWS Kinesis 与 Lambda 处理流数据

speederAggregator 回调函数

import boto3, pytz, pandas as pd

s3 = boto3.client("s3" ...)
...

def speeder_aggregator(event, context):
    tz = pytz.timezone('America/Los_Angeles')
    filter_date = datetime.now(tz).strftime("%Y%m%d")
使用 AWS Kinesis 与 Lambda 处理流数据

speederAggregator 回调函数

def speeder_aggregator(event, context):
  ...
  objects = s3.list_objects_v2(
      Bucket='sd-vehicle-data', Prefix=f'speeders/{filter_date}')
  day_data = []


  for obj in objects['Contents']:
      print(obj['Key'])
      day_record = s3.get_object(Bucket='sd-vehicle-data', Key = obj['Key'])
      day_data.append(
          pd.read_csv(day_record['Body'], delimiter = " "))

  # 将新记录合并为一个数据框
  data = pd.concat(day_data)
  data.columns = ["record_id", "timestamp", "vin", "lon", "lat", "speed"]
使用 AWS Kinesis 与 Lambda 处理流数据

awswrangler 包

import awswrangler as wr

session = boto3.Session(aws_access_key_id = AWS_KEY, 
                        aws_secret_access_key = AWS_SECRET, region_name="us-east-1")

df = wr.s3.read_csv(f"s3://sd-vehicle-data/speeders/{filter_date}")      
df.head()
使用 AWS Kinesis 与 Lambda 处理流数据

旧方法 vs awswrangler

def speeder_aggregator(event, context):
  ...
  objects = s3.list_objects_v2(
      Bucket='sd-vehicle-data', Prefix=f'speeders/{filter_date}')
  day_data = []

  for obj in objects['Contents']:
      print(obj['Key'])
      day_record = s3.get_object(Bucket='sd-vehicle-data', Key = obj['Key'])
      day_data.append(
          pd.read_csv(day_record['Body'], delimiter = " "))

  # 将新记录合并为一个数据框
  data = pd.concat(day_data)
  data.columns = ["record_id", "timestamp", "vin", "lon", "lat", "speed"]
使用 AWS Kinesis 与 Lambda 处理流数据

旧方法 vs awswrangler

import awswrangler as wr

session = boto3.Session(aws_access_key_id = AWS_KEY, 
                        aws_secret_access_key = AWS_SECRET, region_name="us-east-1")
def speeder_aggregator(event, context):
    df = wr.s3.read_csv(f"s3://sd-vehicle-data/speeders/{filter_date}")      
    df.head()
使用 AWS Kinesis 与 Lambda 处理流数据

写入汇总超速文件

wr.s3.to_csv(df, 
             f"s3://sd-vehicle-data/speeders_daily/{filter_date}.csv", 
             boto3_session=session, 
             sep=" ", index=False)
使用 AWS Kinesis 与 Lambda 处理流数据

保存并测试

保存并测试

使用 AWS Kinesis 与 Lambda 处理流数据

启用触发器

启用触发器

使用 AWS Kinesis 与 Lambda 处理流数据

回顾

回顾

使用 AWS Kinesis 与 Lambda 处理流数据

回顾

回顾

使用 AWS Kinesis 与 Lambda 处理流数据

回顾

  • 用 Cron 进行调度
  • wr.s3.read_csv()
  • wr.s3.write_csv()
使用 AWS Kinesis 与 Lambda 处理流数据

Passons à la pratique !

使用 AWS Kinesis 与 Lambda 处理流数据

Preparing Video For Download...