Lambda transformasional

Streaming Data dengan AWS Kinesis dan Lambda

Maksim Pecherskiy

Data Engineer

Di bab 1...

Di bab 2...

Di bab ini...

Pendekatan sebelumnya

Pemrosesan sekali di S3 vs transformasi Lambda

Pemrosesan sekali di S3

Menggunakan fungsi Lambda yang dipicu saat objek ditulis di S3
Jeda lebih lama hingga data bisa ditransformasi
Data mentah disimpan sebelum dibersihkan
Firehose harus ke tujuan S3

Pemrosesan via transformasi Lambda

Menggunakan fungsi Lambda yang dipicu di tengah aliran Firehose
Transformasi langsung terjadi
Hanya data bersih yang disimpan
Memungkinkan tujuan lain di Firehose

Data masuk

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

Lambda transformasional

Contoh event

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

Memroses data

import base64
from datetime import datetime as dt

def convert_timestamp(record_time_val):

    # Dapatkan tanggal hari ini sebagai string
    today = dt.today().strftime("%Y-%m-%d")

    # Gabungkan tanggal hari ini dengan waktu pada record dan buat objek datetime
    new_ts = dt.strptime(f"{today} {record_time_val}", "%Y-%m-%d %H:%M:%S.%f")

    # Ubah objek datetime menjadi string berformat rapi
    return new_ts.strftime("%Y-%m-%dT%H:%M:%S")

Memroses data

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

Memroses data

    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}

Tinjauan singkat

import base64
def lambda_handler(event, context):

    output = []
    for record in event['records']: # Iterasi tiap record

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

        # Ubah isinya

        payload_enc = base64.b64encode(payload.encode()) # Enkode ulang

        output.append({ # Masukkan ke dictionary
            'recordId': record['recordId'], 'result': 'Ok', 'data': payload_enc,
        })

    return {'records': output}

Membuat Lambda di AWS

Skrip untuk membuat Lambda

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

Ayo berlatih!

Streaming Data dengan AWS Kinesis dan Lambda