AWS Kinesis と Lambda で学ぶストリーミングデータ
Maksim Pecherskiy
Data Engineer
{ "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==" } ] }
{ "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=="
} ] }
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") # 見やすい文字列に変換 return new_ts.strftime("%Y-%m-%dT%H:%M:%S")
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")
# 見やすい文字列に変換 return new_ts.strftime("%Y-%m-%dT%H:%M:%S")
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])
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])
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}
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}
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}
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, })
https://drive.google.com/file/d/1fTzIp2mgWt04nVdILc7ClyAMOF1yNxkv/view?usp=sharing
Preparing Video For Download...
