Een transformerende Lambda

Streaming data met AWS Kinesis en Lambda

Maksim Pecherskiy

Data Engineer

In hoofdstuk 1…

In hoofdstuk 2…

In dit hoofdstuk…

Eerdere aanpak

Eén keer verwerken in S3 vs Lambda-transform

Eén keer verwerken in S3

Gebruikt een Lambda-functie die triggert bij schrijven in S3
Langere vertraging tot transformatie
Ruwe data wordt opgeslagen vóór opschonen
Vereist S3 als Firehose-doel

Verwerken via Lambda-transform

Gebruikt een Lambda-functie midden in de Firehose-stream
Transformatie is direct
Alleen opgeschoonde data wordt opgeslagen
Maakt andere Firehose-doelen mogelijk

Binnenkomende data

record_id	timestamp	vin	lon	lat	speed
939ed1d1-1740-420c-8906-445278573c7f	4:25:06.000	4FTEX4944AK844294	106.9447146	-6.3385652	25
f29a5b3d-d0fa-43c0-9e1a-e2a5cdb8be7a	8:10:47.000	3FTEX1G5XAK844393	108.580681	34.79925	37
ff8e7131-408d-463b-8d07-d016419b0656	20:26:44.000	2LAXX1C8XAK844292	114.392392	36.097577	90

Transformerende Lambda

Voorbeeldevent

{
  "invocationId": "invocationIdExample",
  "deliveryStreamArn": "arn:aws:firehose:us-east-1:458912630:deliverystream/gps-delivery-stream",
  "region": "us-east-1",
  "records": [
    {

      "recordId": "49546986683135544286507457936321625675700192471156785154",

      "approximateArrivalTimestamp": 1495072949453,

      "data": "NjQuMjQyLjg4LjEwIC0gLSBbMDcvTWFyLzIwMDQ6MTY6MTA6MDIgLTA4MDBdICJHRVQgL21haWxtYW4vbGlzdGluZm8vaHNkaXZpc2lvbiBIVFRQLzEuMSIgMjAwIDYyOTE=="

    }
  ]
}

Base64

De data verwerken

import base64
from datetime import datetime as dt

def convert_timestamp(record_time_val):

    # Haal de datum van vandaag als string op
    today = dt.today().strftime("%Y-%m-%d")

    # Combineer met de tijd uit het record en maak een datetime-object
    new_ts = dt.strptime(f"{today} {record_time_val}", "%Y-%m-%d %H:%M:%S.%f")

    # Geef het terug als netjes geformatteerde string
    return new_ts.strftime("%Y-%m-%dT%H:%M:%S")

De data verwerken

def lambda_handler(event, context):
    output = []

    for record in event['records']:
        payload = base64.b64decode(record['data'])

        payload = payload.decode()

        payload = payload.split(" ")

        payload[1] = convert_timestamp(payload[1])

De data verwerken

    for record in event['records']:
        ...
        payload = " ".join(payload)
        payload_enc = base64.b64encode(payload.encode())

        output.append({

          'recordId': record['recordId'],

          'result': 'Ok',

          'data': payload_enc

        })

    return {'records': output}

Een herhaling

import base64
def lambda_handler(event, context):

    output = []
    for record in event['records']: # Loop over de records

        payload = base64.b64decode(record['data']).decode() # Decode de payload

        # Pas aan

        payload_enc = base64.b64encode(payload.encode()) # Encode opnieuw

        output.append({ # Voeg toe aan dict
            'recordId': record['recordId'], 'result': 'Ok', 'data': payload_enc,
        })

    return {'records': output}

Lambda maken in AWS

Script om Lambda te maken

https://drive.google.com/file/d/1fTzIp2mgWt04nVdILc7ClyAMOF1yNxkv/view?usp=sharing

Laten we oefenen!

Streaming data met AWS Kinesis en Lambda