Quy trình dữ liệu serverless

Xử lý dữ liệu streaming với AWS Kinesis và Lambda

Maksim Pecherskiy

Data Engineer

Ôn lại

Phân vùng hiện tại

Lọc người vượt tốc

Tổng hợp theo ngày

Mở recordReaderS3

Chỉnh sửa recordReaderS3

...
import pytz
from datetime import datetime
tz = pytz.timezone('America/Los_Angeles')

Chỉnh sửa recordReaderS3

def record_created_handler(event, context):

    ...
    ## Lấy các tốc độ vượt quá giới hạn 45
    too_fast = top_speeds.loc[top_speeds.speed > SPEED_ALERT_THRESHOLD :]

    ## Tạo khóa đối tượng
    fdate = datetime.now(tz).strftime("%Y%m%d/%HH%MM%SS")
    obj_key = f"speeders/{fdate}.csv" # tên tệp trong thư mục speeders


    ## Ghi đối tượng lên S3
    s3.put_object(Bucket='sd-vehicle-data', 
                  Key=obj_key,
                  Body=too_fast.to_csv(sep=" ", index=False)
    )

Ghi người vượt tốc theo ngày, giờ

speederAggregator

Tạo speederAggregator

Thêm layer AWS Data Wrangler

Tài nguyên speederAggregator

Tạo trigger theo thời gian

cron(50 6 * * ? *)

Vào phút 50
Của giờ thứ 6 (UTC)
Mỗi ngày trong tháng
Mỗi tháng
Mỗi ngày trong tuần
Mỗi năm

Hàm gọi lại 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")

Hàm gọi lại 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 = " "))

  # Gộp bản ghi mới thành một dataframe.
  data = pd.concat(day_data)
  data.columns = ["record_id", "timestamp", "vin", "lon", "lat", "speed"]

Gói 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()

Cách cũ 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 = " "))

  # Gộp bản ghi mới thành một dataframe.
  data = pd.concat(day_data)
  data.columns = ["record_id", "timestamp", "vin", "lon", "lat", "speed"]

Cách cũ 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()

Ghi tệp tổng hợp người vượt tốc

wr.s3.to_csv(df, 
             f"s3://sd-vehicle-data/speeders_daily/{filter_date}.csv", 
             boto3_session=session, 
             sep=" ", index=False)

Quy trình dữ liệu serverless

Ôn lại

Phân vùng hiện tại

Lọc người vượt tốc

Tổng hợp theo ngày

Mở recordReaderS3

Chỉnh sửa recordReaderS3

Chỉnh sửa recordReaderS3

Ghi người vượt tốc theo ngày, giờ

speederAggregator

Tạo speederAggregator

Thêm layer AWS Data Wrangler

Tài nguyên speederAggregator

Tạo trigger theo thời gian

Hàm gọi lại speederAggregator

Hàm gọi lại speederAggregator

Gói awswrangler

Cách cũ vs awswrangler

Cách cũ vs awswrangler

Ghi tệp tổng hợp người vượt tốc

Lưu và kiểm thử

Bật trigger

Tổng kết

Tổng kết

Tổng kết

Ayo berlatih!