Dönüştürücü Lambda

AWS Kinesis ve Lambda ile Akış Verileri

Maksim Pecherskiy

Data Engineer

Bölüm 1'de...

Bölüm 2'de...

Bu bölümde...

Önceki yaklaşım

S3'te işleme vs Lambda dönüşümü

S3'te tek seferde işleme

S3'e yazmada tetiklenen bir Lambda kullanır
Dönüşüm için daha uzun gecikme
Ham veri temizlikten önce saklanır
Firehose hedefinin S3 olmasını gerektirir

Lambda dönüşümü ile işleme

Firehose akışı ortasında tetiklenen bir Lambda kullanır
Dönüşüm anlıktır
Yalnızca temiz veri saklanır
Firehose'da başka hedeflere izin verir

Gelen veriler

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

Dönüştürücü Lambda

Örnek olay

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

Veriyi işleme

import base64
from datetime import datetime as dt

def convert_timestamp(record_time_val):

    # Bugünün tarihini metin olarak al
    today = dt.today().strftime("%Y-%m-%d")

    # Bugünün tarihi ile kaydın saatini birleştir ve datetime nesnesi yap
    new_ts = dt.strptime(f"{today} {record_time_val}", "%Y-%m-%d %H:%M:%S.%f")

    # Datetime nesnesini düzgün biçimlendirilmiş metne çevir
    return new_ts.strftime("%Y-%m-%dT%H:%M:%S")

Veriyi işleme

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

Veriyi işleme

    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}

Kısa özet

import base64
def lambda_handler(event, context):

    output = []
    for record in event['records']: # Kayıtlar üzerinde yinele

        payload = base64.b64decode(record['data']).decode() # Yükü çöz

        # Değiştir

        payload_enc = base64.b64encode(payload.encode()) # Yeniden kodla

        output.append({ # Sözlüğe ekle
            'recordId': record['recordId'], 'result': 'Ok', 'data': payload_enc,
        })

    return {'records': output}

AWS'de Lambda oluşturma

Lambda oluşturma betiği

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

Hadi pratik yapalım!

AWS Kinesis ve Lambda ile Akış Verileri