सर्वरलेस डेटा वर्कफ़्लो

AWS Kinesis और Lambda के साथ स्ट्रीमिंग डेटा

Maksim Pecherskiy

Data Engineer

एक नज़र पीछे

वर्तमान पार्टिशनिंग

स्पीडर फ़िल्टर करना

दिन के अनुसार एग्रीगेट करना

recordReaderS3 खोलें

recordReaderS3 संपादित करना

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

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

तारीख-समय के अनुसार स्पीडर लिखना

speederAggregator

speederAggregator बनाएँ

AWS Data Wrangler लेयर जोड़ें

speederAggregator संसाधन

टाइम्ड ट्रिगर बनाएँ

cron(50 6 * * ? *)

Minute 50 पर
6वें घंटे (UTC) पर
हर महीने के हर दिन
हर महीने
हफ्ते के हर दिन
हर साल

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

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

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

पुराना तरीका बनाम 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"]

पुराना तरीका बनाम 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()

एग्रीगेटेड स्पीडर फ़ाइल लिखना

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

सेव करें और टेस्ट करें

ट्रिगर सक्षम करें

समीक्षा

शेड्यूलिंग के लिए Cron
wr.s3.read_csv()
wr.s3.write_csv()

अभ्यास करते हैं!

AWS Kinesis और Lambda के साथ स्ट्रीमिंग डेटा