Flux de date serverless

Date în flux cu AWS Kinesis și Lambda

Maksim Pecherskiy

Data Engineer

Privire de ansamblu

z8HRk.png

Date în flux cu AWS Kinesis și Lambda

Partiționarea curentă

FMktI.png

Date în flux cu AWS Kinesis și Lambda

Filtrarea depășirilor de viteză

IPwGK.png

Date în flux cu AWS Kinesis și Lambda

Agregare pe zi

tvBXx.png

Date în flux cu AWS Kinesis și Lambda

Deschiderea recordReaderS3

aIahM.png

Date în flux cu AWS Kinesis și Lambda

Editarea recordReaderS3

...
import pytz
from datetime import datetime
tz = pytz.timezone('America/Los_Angeles')
Date în flux cu AWS Kinesis și Lambda

Editarea recordReaderS3

def record_created_handler(event, context):

    ...
    ## Get top speeds that exceed the limit of 45
    too_fast = top_speeds.loc[top_speeds.speed > SPEED_ALERT_THRESHOLD :]

    ## Generate object key
    fdate = datetime.now(tz).strftime("%Y%m%d/%HH%MM%SS")
    obj_key = f"speeders/{fdate}.csv" # filename in speeders folder


    ## Write the object to S3
    s3.put_object(Bucket='sd-vehicle-data', 
                  Key=obj_key,
                  Body=too_fast.to_csv(sep=" ", index=False)
    )
Date în flux cu AWS Kinesis și Lambda

Scrierea depășirilor după dată și oră

s3QY4.png

Date în flux cu AWS Kinesis și Lambda

speederAggregator

tvBXx.png

Date în flux cu AWS Kinesis și Lambda

Crearea speederAggregator

3pGYX.png

Date în flux cu AWS Kinesis și Lambda

Adăugarea stratului AWS Data Wrangler

7TO0V.png

Date în flux cu AWS Kinesis și Lambda

Resurse speederAggregator

sO0op.png

Date în flux cu AWS Kinesis și Lambda

Crearea declanșatorului programat

4Qi4i.png

Date în flux cu AWS Kinesis și Lambda

y0FkQ.png

cron(50 6 * * ? *)
  • La minutul 50
  • Din ora a 6-a (UTC)
  • Din fiecare zi a lunii
  • Din fiecare lună
  • Din fiecare zi a săptămânii
  • Din fiecare an
Date în flux cu AWS Kinesis și Lambda

Callback-ul 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")
Date în flux cu AWS Kinesis și Lambda

Callback-ul 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 = " "))

  # Concatenate new records into a single dataframe.
  data = pd.concat(day_data)
  data.columns = ["record_id", "timestamp", "vin", "lon", "lat", "speed"]
Date în flux cu AWS Kinesis și Lambda

Pachetul 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()
Date în flux cu AWS Kinesis și Lambda

Metoda clasică 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 = " "))

  # Concatenate new records into a single dataframe.
  data = pd.concat(day_data)
  data.columns = ["record_id", "timestamp", "vin", "lon", "lat", "speed"]
Date în flux cu AWS Kinesis și Lambda

Metoda clasică 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()
Date în flux cu AWS Kinesis și Lambda

Scrierea fișierului agregat de depășiri

wr.s3.to_csv(df, 
             f"s3://sd-vehicle-data/speeders_daily/{filter_date}.csv", 
             boto3_session=session, 
             sep=" ", index=False)
Date în flux cu AWS Kinesis și Lambda

Salvare și testare

agF2b.png

Date în flux cu AWS Kinesis și Lambda

Activarea declanșatorului

COfYY.png

Date în flux cu AWS Kinesis și Lambda

Recapitulare

u1Jwm.png

Date în flux cu AWS Kinesis și Lambda

Recapitulare

7ZYPV.png

Date în flux cu AWS Kinesis și Lambda

Recapitulare

  • Cron pentru planificare
  • wr.s3.read_csv()
  • wr.s3.write_csv()
Date în flux cu AWS Kinesis și Lambda

Să exersăm!

Date în flux cu AWS Kinesis și Lambda

Preparing Video For Download...