Sunucusuz veri iş akışı

AWS Kinesis ve Lambda ile Akış Verileri

Maksim Pecherskiy

Data Engineer

Geriye bakış

Geçerli bölümleme

Hız ihlalcilerini filtreleme

Güne göre toplama

recordReaderS3'ü açın

recordReaderS3'ü düzenleme

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

recordReaderS3'ü düzenleme

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

Hız ihlallerini tarih ve saate göre yazma

speederAggregator

speederAggregator oluşturma

AWS data wrangler katmanı ekleme

speederAggregator Kaynakları

Zamanlanmış tetikleyici oluşturma

cron(50 6 * * ? *)

1. dakikada
1. saatte (UTC)
Ayın her günü
Her ay
Haftanın her günü
Her yıl

speederAggregator geri çağrısı

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 geri çağrısı

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 paketi

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

Eski yöntem 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"]

Eski yöntem 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()

Toplu hız ihlali dosyasını yazma

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

Sunucusuz veri iş akışı

Geriye bakış

Geçerli bölümleme

Hız ihlalcilerini filtreleme

Güne göre toplama

recordReaderS3'ü açın

recordReaderS3'ü düzenleme

recordReaderS3'ü düzenleme

Hız ihlallerini tarih ve saate göre yazma

speederAggregator

speederAggregator oluşturma

AWS data wrangler katmanı ekleme

speederAggregator Kaynakları

Zamanlanmış tetikleyici oluşturma

speederAggregator geri çağrısı

speederAggregator geri çağrısı

awswrangler paketi

Eski yöntem vs awswrangler

Eski yöntem vs awswrangler

Toplu hız ihlali dosyasını yazma

Kaydet ve test et

Tetikleyiciyi etkinleştirme

Gözden geçirme

Gözden geçirme

Gözden geçirme

Vamos praticar!