转换型 Lambda

使用 AWS Kinesis 与 Lambda 处理流数据

Maksim Pecherskiy

Data Engineer

第 1 章回顾…

2020-07-15_06-18.png

使用 AWS Kinesis 与 Lambda 处理流数据

第 2 章回顾…

2020-07-15_06-20.png

使用 AWS Kinesis 与 Lambda 处理流数据

本章内容…

2020-07-15_06-27.png

使用 AWS Kinesis 与 Lambda 处理流数据

先前的方法

2020-07-14_07-42.png

使用 AWS Kinesis 与 Lambda 处理流数据

S3 一次性处理 vs Lambda 转换

在 S3 中一次性处理

  • 使用在 S3 写入时触发的 Lambda 函数
  • 数据转换延迟更长
  • 先存原始数据，再清洗
  • 需将 Firehose 目标设为 S3

通过 Lambda 转换处理

  • 使用在 Firehose 流中途触发的 Lambda 函数
  • 转换即时完成
  • 仅存储清洗后的数据
  • 允许 Firehose 使用其他目标
使用 AWS Kinesis 与 Lambda 处理流数据

传入数据

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
使用 AWS Kinesis 与 Lambda 处理流数据

转换型 Lambda

2020-07-15_06-38.png

使用 AWS Kinesis 与 Lambda 处理流数据

示例事件

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

    }
  ]
}
使用 AWS Kinesis 与 Lambda 处理流数据

Base64

2020-07-19_10-54.png

使用 AWS Kinesis 与 Lambda 处理流数据

处理数据

import base64
from datetime import datetime as dt

def convert_timestamp(record_time_val):

    # 获取今天日期为字符串
    today = dt.today().strftime("%Y-%m-%d")

    # 将今天日期与记录时间合并并创建 datetime 对象
    new_ts = dt.strptime(f"{today} {record_time_val}", "%Y-%m-%d %H:%M:%S.%f")

    # 将 datetime 转为格式化字符串
    return new_ts.strftime("%Y-%m-%dT%H:%M:%S")
使用 AWS Kinesis 与 Lambda 处理流数据

处理数据

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])
使用 AWS Kinesis 与 Lambda 处理流数据

处理数据

    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}
使用 AWS Kinesis 与 Lambda 处理流数据

回顾

import base64
def lambda_handler(event, context):

    output = []
    for record in event['records']: # 遍历记录

        payload = base64.b64decode(record['data']).decode() # 解码负载

        # 修改内容

        payload_enc = base64.b64encode(payload.encode()) # 重新编码

        output.append({ # 放入字典
            'recordId': record['recordId'], 'result': 'Ok', 'data': payload_enc,
        })

    return {'records': output}
使用 AWS Kinesis 与 Lambda 处理流数据

在 AWS 中创建 Lambda

使用 AWS Kinesis 与 Lambda 处理流数据

创建 Lambda 的脚本

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

使用 AWS Kinesis 与 Lambda 处理流数据

¡Vamos a practicar!

使用 AWS Kinesis 与 Lambda 处理流数据

Preparing Video For Download...