Serverloze dataworkflow

Streaming data met AWS Kinesis en Lambda

Maksim Pecherskiy

Data Engineer

Even terugkijken

Huidige partitionering

Speeders eruit filteren

Per dag aggregeren

recordReaderS3 openen

recordReaderS3 bewerken

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

recordReaderS3 bewerken

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)
    )

Speeders schrijven per datum en tijd

speederAggregator

speederAggregator maken

AWS Data Wrangler-laag toevoegen

speederAggregator-resources

De getimede trigger maken

cron(50 6 * * ? *)

Op minuut 50
Van het 6e uur (UTC)
Elke dag van de maand
Elke maand
Elke dag van de week
Elk jaar

speederAggregator-callback

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")

speederAggregator-callback

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"]

awswrangler-pakket

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()

Oude manier 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"]

Oude manier 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()

Samengevoegd speeders-bestand schrijven

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

Serverloze dataworkflow

Even terugkijken

Huidige partitionering

Speeders eruit filteren

Per dag aggregeren

recordReaderS3 openen

recordReaderS3 bewerken

recordReaderS3 bewerken

Speeders schrijven per datum en tijd

speederAggregator

speederAggregator maken

AWS Data Wrangler-laag toevoegen

speederAggregator-resources

De getimede trigger maken

speederAggregator-callback

speederAggregator-callback

awswrangler-pakket

Oude manier vs awswrangler

Oude manier vs awswrangler

Samengevoegd speeders-bestand schrijven

Opslaan en testen

Trigger inschakelen

Herhaling

Herhaling

Herhaling

Laten we oefenen!