Pada bab pertama ini, Anda akan mempelajari perbedaan antara pemrosesan batch dan stream, membuat stream pertama Anda, mengelola izinnya, menulis ke stream, dan membacanya.

Apa itu streaming dan mengapa hal itu penting?

Batch vs stream

Produser dan destination

Mengelola Firehose delivery stream

Bersiap untuk stream pertama

Membuat role

Perbandingan IAM users vs. roles

Latihan izin akses

Pembuatan bucket S3

Bekerja dengan delivery stream Firehose

Buat Firehose stream pertama Anda

Menulis ke stream Firehose

Membaca data Firehose

Streaming di cloud

Langkah berikutnya dalam perjalanan data streaming Anda adalah mempelajari cara menggunakan fungsi lambda transformasional untuk beralih ke arsitektur tanpa server. Melalui latihan praktik, Anda akan menambahkan lapisan lambda dan memicu fungsi lambda berdasarkan kondisi tertentu.

Menuju serverless

Membuat dan menjalankan fungsi Lambda

Opsi

Cocok untuk Server

Cocok untuk Serverless

Serverless vs server

Trigger

Handler

Layer

Komponen fungsi Lambda

# EDIT ME!
# recordReaderS3/lambda_function.py
# This is lambda handler code that will execute each time an object is added to S3.
import json

def record_created_handler(____, ____):
    # Create a list to store new object keys
    written_objects = []

    # Iterate over each S3 event record
    for record in event['Records']:
       
        # Get the variables to check for
        event_name = record['eventName']
        bucket_name = record['s3']['bucket']['name']
        obj_key = record['s3']['object']['key']

        # Verify that the object was written to S3
        if event_name == 'ObjectCreated:____' and bucket_name == 'sd-vehicle-data':
           
            # If so, then add it to the written_objects list
            written_objects.append(obj_key)
    
    # Return the collected written objects
    return {
        "bucket": 'sd-vehicle-data',
        "new_objects": written_objects
    }

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json


# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("recordReaderS3", "record_created_handler")

# Read the test event JSON file as dictionary
with open("./dcs/sample_events/s3_event_c2l1.json") as json_file:
    payload = json.load(json_file)

# Invoke lambda with test event, print result
result = invoke_function_and_get_message("recordReaderS3", payload)
print(json.dumps(json.loads(result), indent=4))

# EDIT ME!
# recordReaderS3/lambda_function.py
# This is lambda handler code that will execute each time an object is added to S3
import json

def record_created_handler(event, context):
    # Create a list to store new object keys
    written_objects = []

    # Iterate over each S3 event record
    for record in event['Records']:
       
        # Get the variables to check for
        event_name = record['eventName']
        bucket_name = record['s3']['bucket']['name']
        obj_key = record['s3']['object']['key']

        # Verify that the object was written to S3
        if event_name == 'ObjectCreated:Put' and bucket_name == 'sd-vehicle-data':
           
            # If so, then add it to the written_objects list
            written_objects.append(obj_key)
    
    # Return the collected written objects
    return {
        "bucket": 'sd-vehicle-data',
        "new_objects": written_objects
    }                 

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event
from dcs.util.aws_lambda import * 
import json


# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("recordReaderS3", "record_created_handler")

# Read the test event JSON file as dictionary
with open("./dcs/sample_events/s3_event_c2l1.json") as json_file:
    payload = json.load(json_file)

# Invoke lambda with test event, print result
result = invoke_function_and_get_message("recordReaderS3", payload)
print(json.dumps(json.loads(result), indent=4))

Membaca data dari S3 berdasarkan sebuah peristiwa

Lambda live pertama Anda!

Menambahkan lapisan lambda

Alur Lambda

Menggunakan variabel lingkungan

Alur kerja data tanpa server

Lambda yang berjalan cepat

Lambda yang membutuhkan waktu lebih lama untuk berjalan

Lambda cepat vs lambat

Membuat Time-Triggered Lambda

Lambda yang dipicu oleh waktu

Kode lambda berbasis jadwal waktu

API tanpa server

Buat Lambda yang Dipicu API

Buat Lambda API

import boto3

lambda_client = boto3.client('lambda', aws_access_key_id="12345",
    aws_secret_access_key="12345",
    region_name='us-east-1',
    endpoint_url="http://localhost:4574")

def trigger_full_report_builder(event, context):
    async = event['____']['async']
    if async == 'true':
        exec_type = '____'
    else:
        exec_type = 'RequestResponse'
        
    result = lambda_client.____(
      FunctionName='fullReportBuilder',
      InvocationType=exec_type)
    
    return {
        'statusCode': 200,
        'headers': {
            "content-type" : "application/json"
        },
        'body': "Successfully Invoked"
    }

import awswrangler_c as wr
import boto3
# Initialize boto3 session
session = boto3.Session(aws_access_key_id = "12345", 
  aws_secret_access_key = "12345", 
  region_name="us-east-1")

def full_report_builder(event, context):
  # Read all records in the speeders folder
  speeders_total = wr.s3.read_csv('s3://sd-vehicle-data/speeders', 
                                  boto3_session = session, 
                                  delimiter=" ")

  # Write aggregated speeders file
  return wr.s3.to_csv(df = speeders_total, 
               boto3_session=session, 
               path="s3://sd-vehicle-data/speeders-full/full.csv")
               

# PLEASE DO NOT EDIT
# aws_wrangler_c.py
def read_csv(path, boto3_session, delimiter):
  bucket = path.split("//")[1].split("/")[0]
  prefix = path.split("//")[1].replace(f"{bucket}/", "")
  objects = s3.list_objects_v2(Bucket=bucket, Prefix=prefix)['Contents']
  dfs = []
  for o in objects:
      key = o['Key']
      data = s3.get_object(Bucket='sd-vehicle-data', Key=key)
      dfs.append(pd.read_csv(data['Body'], delimiter=delimiter))
  return pd.concat(dfs)
  
def to_csv(df, path, boto3_session, separator=","):
  bucket = path.split("//")[1].split("/")[0]
  key = path.split("//")[1].replace(f"{bucket}/", "")
  s3.put_object(Bucket=bucket, Key=key, Body=df.to_csv(sep=separator, index=False))
  return True

s3 = {
    "to_csv": to_csv,
    "read_csv": read_csv
}
                                                  

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json

# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("fullReportBuilder", "full_report_builder")
create_lambda("triggerFullReportBuilder", "trigger_full_report_builder")


# Invoke lambda with test event, print result.
result = invoke_function_and_get_message("triggerFullReportBuilder", {
  "queryStringParameters": {
    "async": "true"
  },
  "isBase64Encoded": "false"
})
print(result)

import boto3

lambda_client = boto3.client('lambda', aws_access_key_id="12345",
    aws_secret_access_key="12345",
    region_name='us-east-1',
    endpoint_url="http://localhost:4574")

def trigger_full_report_builder(event, context):
    async = event['queryStringParameters']['async']
    if async == 'true':
        exec_type = 'Event'
    else:
        exec_type = 'RequestResponse'
        
    result = lambda_client.invoke(
      FunctionName='fullReportBuilder',
      InvocationType=exec_type)
    
    return {
        'statusCode': 200,
        'headers': {
            "content-type" : "application/json"
        },
        'body': "Successfully Invoked"
    }

    

import awswrangler_c as wr
import boto3
# Initialize boto3 session
session = boto3.Session(aws_access_key_id = "12345", 
  aws_secret_access_key = "12345", 
  region_name="us-east-1")

def full_report_builder(event, context):
  # Read all records in the speeders folder
  speeders_total = wr.s3.read_csv('s3://sd-vehicle-data/speeders', 
                                  boto3_session = session, 
                                  delimiter=" ")

  # Write aggregated speeders file
  return wr.s3.to_csv(df = speeders_total, 
               boto3_session=session, 
               path="s3://sd-vehicle-data/speeders-full/full.csv")
               
  

# PLEASE DO NOT EDIT
# run_lambda.py
# This executes a lambda function with a sample event.
from dcs.util.aws_lambda import * 
import json

# Clean up any old lambda functions
clean_lambdas()

# Create a new lambda 
create_lambda("fullReportBuilder", "full_report_builder")
create_lambda("triggerFullReportBuilder", "trigger_full_report_builder")

# Invoke lambda with test event, print result.
result = invoke_function_and_get_message("triggerFullReportBuilder", {
  "queryStringParameters": {
    "async": "true"
  },
  "isBase64Encoded": "false"
})
print(result)

Menuju tanpa server

Sekarang Anda siap untuk melakukan encoding dan decoding data streaming serta menganalisis data langsung di dalam stream. Anda bahkan akan menggunakan beberapa stream untuk mendapatkan kecepatan puncak harian kendaraan.

Lambda yang transformatif

Mentransformasikan data di dalam sebuah stream

Menyandi dan mengurai base64

Buat Lambda transformasional

Sebuah lambda transformasi sederhana

Menganalisis data dalam stream

Membuat aplikasi analitik data Kinesis

Kinesis Data Analytics vs Transformational Lambda

Membangun aplikasi analitik data Kinesis

Komponen SQL Kinesis Data Analytics

Menggunakan beberapa stream

Menyampaikan data dari Kinesis Analytics

Dapatkan kecepatan puncak harian

Menggunakan Kinesis Data Analytics untuk peringatan

Menganalisis data streaming

Di bab terakhir ini, Anda akan mempelajari cara memantau kinerja stream menggunakan log, metrik, alarm, dan dasbor. Anda akan menggunakan Elasticsearch dan membangun dasbor Kibana Anda sendiri.

Studi kasus data streaming

Membuat klaster Elasticsearch

RedShift vs Elasticsearch

Memantau kinerja

Dasbor dan alarm CloudWatch

Komponen CloudWatch

Alur pemantauan CloudWatch

Memvisualisasikan data streaming

Bekerja dengan ElasticSearch menggunakan Kibana

CloudWatch vs Elasticsearch

Alur visualisasi Elasticsearch

Pendekatan alternatif

Pendekatan alternatif lainnya

Memantau dan memvisualisasikan data streaming

Apa yang menggerakkan sistem yang kita gunakan setiap hari tanpa kita sadari, seperti deteksi penipuan untuk menjaga keamanan transaksi atau lampu lalu lintas yang membuat arus kendaraan tetap lancar? Jawabannya adalah data streaming (data yang dihasilkan terus-menerus dari berbagai sumber) dan teknologi tanpa server—seperti Amazon Kinesis dan AWS Lambda.
<br /> <br />
Dalam kursus ini, Anda akan mempelajari cara memanfaatkan teknologi kuat tersebut dengan membantu seorang data engineer fiksi bernama Cody. Tujuan Anda adalah membantunya mengumpulkan data streaming waktu nyata dari kendaraan milik kota, menganalisis data tersebut, dan mengirim peringatan yang relevan seperti peringatan kecepatan kepada pengemudi. Dengan Amazon Kinesis dan Firehose, Anda akan belajar cara mengumpulkan data dari jutaan sumber sebelum menggunakan Kinesis Analytics untuk menganalisis data saat data mengalir dalam stream. Anda juga akan membuat fungsi tanpa server di AWS Lambda yang akan memicu tindakan secara kondisional berdasarkan data yang diterima. Pada akhir pelatihan ini Anda akan mengetahui cara membuat dasbor ElasticSearch langsung dengan AWS QuickSight dan CloudWatch—dan semoga membantu Cody menyelesaikan proyek ambisiusnya.


Introduction to AWS Boto in Python

Introduction to Shell

Streaming Concepts

Pelajari cara bekerja dengan data streaming menggunakan teknologi serverless di AWS.

Streaming Data dengan AWS Kinesis dan Lambda

Cocok untuk memicu fungsi lain secara asinkron

Memerlukan batas waktu eksekusi yang lebih rendah

Digunakan untuk membaca banyak file dan mengagregasikannya

Dapat bergantung pada hasil sinkron dari menjalankan lambda lain

Mungkin memerlukan batas waktu eksekusi yang lebih tinggi

Lambda cepat vs lambat

Streaming Data dengan AWS Kinesis dan Lambda

Latihan interaktif langsung