In this first chapter, you will learn about the differences between batch and stream processing, create your first stream, manage its permissions, write to it and read from it.

Apa itu streaming dan mengapa hal itu penting?

Options

Batch

Stream

Batch vs stream

Opsi

Data Producers

..memasukkan data ke Firehose Stream

...yang mengirimkannya ke Destination

Produser dan destination

Mengelola Firehose delivery stream

Bersiap untuk stream pertama

Membuat role

IAM User

Role

Baik IAM Users maupun Roles

Perbandingan IAM users vs. roles

Pilihan

Permissions for User only

Permissions for User and Role

Latihan izin akses

Pembuatan bucket S3

Bekerja dengan delivery stream Firehose

# _setup.py: Create clients, load data, etc. No need to edit.

import boto3

# Create firehose client
firehose = boto3.client('firehose', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4573")

# Create s3 client
s3 = boto3.client('s3', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4572")

# Prep variables for export
ex_vars = [firehose, s3]


# create_firehose.py: Create firehose stream. No need to edit.
import _setup
firehose, s3 = _setup.ex_vars

# Create s3 bucket
s3.create_bucket(Bucket='sd-vehicle-data')

# Create Firehose delivery stream
res = firehose.____(
    DeliveryStreamName="gps-delivery-stream",
    DeliveryStreamType="DirectPut",
    # Specify configuration of the destination S3 bucket
    ____ = {
        "BucketARN": "arn:aws:s3:::sd-vehicle-data",
        "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole"
    })
    
# Print the stream ARN
print("Created Firehose Stream ARN: {}".format(res['DeliveryStreamARN']))


# _setup.py: Create clients, load data, etc. No need to edit.
import boto3

# Create firehose client
firehose = boto3.client('firehose', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4573")

# Create s3 client
s3 = boto3.client('s3', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4572")

# Prep variables for export
ex_vars = [firehose, s3]


# create_firehose.py: Create firehose stream. No need to edit.
import _setup
firehose, s3 = _setup.ex_vars

# Create s3 bucket
s3.create_bucket(Bucket='sd-vehicle-data')

# Create Firehose delivery stream
res = firehose.create_delivery_stream(
    DeliveryStreamName =  "gps-delivery-stream",
    DeliveryStreamType = "DirectPut",
    
    # Specify configuration of the destination S3 bucket
    S3DestinationConfiguration = {
        "BucketARN": "arn:aws:s3:::sd-vehicle-data",
        "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole"
    })
    
# Print the stream ARN
print("Created Firehose Stream ARN: {}".format(res['DeliveryStreamARN']))


Buat Firehose stream pertama Anda

# _setup.py: Create clients, load data, etc. No need to edit.
import boto3, pandas as pd

# Load Data
data_url = "https://assets.datacamp.com/production/repositories/5668/datasets/6bba555e0e42ae31d1d634256679db718cfb8d76/vehicles.csv"
records = pd.read_csv(data_url).head(100)

# Create firehose client
firehose = boto3.client('firehose', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4573")

# Create s3 client
s3 = boto3.client('s3', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4572")
    
# Prep variables for export
ex_vars = [firehose, s3, records]
    


# create_firehose.py: Create firehose stream. No need to edit.
import _setup
firehose, s3, records = _setup.ex_vars

# Create s3 bucket
s3.create_bucket(Bucket='sd-vehicle-data')

# Create Firehose delivery stream
res = firehose.create_delivery_stream(
    DeliveryStreamName =  "gps-delivery-stream",
    DeliveryStreamType = "DirectPut",
    
    # Specify configuration of the destination S3 bucket
    S3DestinationConfiguration = {
        "BucketARN": "arn:aws:s3:::sd-vehicle-data",
        "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole"
    })
    


# OBD2_sensors.py: Write to Firehose stream. EDIT HERE.
import _setup, create_firehose
firehose, s3, records = _setup.ex_vars
for idx, row in records.iterrows(): 

    # Create a payload string that ends with a newline
    payload = ' '.join(str(value) for value in row) 
    payload = payload + "\n"
    print("Sending payload: {}".format(payload))

    # Send the payload string to Firehose stream
    res = firehose.____(
        DeliveryStreamName = 'gps-delivery-stream',
        Record = {'____': payload})

    # Print the written RecordId
    print("Wrote to RecordId: {}".format(res['RecordId']))


# OBD2_sensors.py: Write to Firehose stream. EDIT HERE.
import _setup, create_firehose
firehose, s3, records = _setup.ex_vars
for idx, row in records.iterrows(): 

    # Create a payload string that ends with a newline
    payload = ' '.join(str(value) for value in row) 
    payload = payload + "\n"
    print("Sending payload: {}".format(payload))

    # Send the payload string to Firehose stream
    res = firehose.put_record(
        DeliveryStreamName = 'gps-delivery-stream',
        Record = {'Data': payload})

    # Print the written RecordId
    print("Wrote to RecordId: {}".format(res['RecordId']))


Menulis ke stream Firehose

# _setup.py: Create clients, load data, etc. No need to edit.
import boto3, pandas as pd

# Load Data
data_url = "https://assets.datacamp.com/production/repositories/5668/datasets/6bba555e0e42ae31d1d634256679db718cfb8d76/vehicles.csv"
records = pd.read_csv(data_url).sample(100)

# Create firehose client
firehose = boto3.client('firehose', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4573")

# Create s3 client
s3 = boto3.client('s3', 
    aws_access_key_id="None", 
    aws_secret_access_key="None", 
    region_name='us-east-1', 
    endpoint_url="http://localhost:4572")
    
# Prep variables for export
ex_vars = [firehose, s3, records]
    


# _run_deps.py: Execute dependencies. No need to edit.
import create_firehose
import OBD2_sensors


# create_firehose.py: Create firehose stream and s3 bucket. No need to edit.
import _setup
firehose, s3, records = _setup.ex_vars

# Create s3 bucket
s3.create_bucket(Bucket='sd-vehicle-data')

# Create Firehose delivery stream
res = firehose.create_delivery_stream(
    DeliveryStreamName =  "gps-delivery-stream",
    DeliveryStreamType = "DirectPut",
    
    # Specify configuration of the destination S3 bucket
    S3DestinationConfiguration = {
        "BucketARN": "arn:aws:s3:::sd-vehicle-data",
        "RoleARN": "arn:aws:iam::0000000:role/firehoseDeliveryRole"
    })
    


# OBD2_sensors.py: Write OBD2 sensor data. No need to edit.
import _setup
firehose, s3, records = _setup.ex_vars
for idx, row in records.iterrows(): 

    # Create a payload string that ends with a newline
    payload = ' '.join(str(value) for value in row) 
    payload = payload + "\n"

    # Send the payload string to Firehose stream
    res = firehose.put_record(
        DeliveryStreamName = 'gps-delivery-stream',
        Record = {'Data': payload})
    


# analyze_data.py: Analyze written sensor data. EDIT HERE.
import _setup, _run_deps, pandas as pd
firehose, s3, records = _setup.ex_vars

# List the objects that have been written to the S3 bucket
objects = s3.____(Bucket='sd-vehicle-data')['Contents']

# Create list for collecting dataframes from read files.
dfs = []

# For every object, load it from S3
for obj in objects:
    data_file = s3.____(Bucket='sd-vehicle-data', ____=obj['Key'])

    # Load it into a dataframe, specifying a delimiter and column names
    dfs.append(pd.read_csv(data_file['Body'], 
                           delimiter = " ", 
                           names=["record_id", "timestamp", "vin", "lon", "lat", "speed"]))

# Concatenate the resulting dataframes.
data = pd.concat(dfs)
print(data.groupby(['vin'])['speed'].max())

# OBD2_sensors.py: Analyze written sensor data. EDIT HERE.
import _setup, _run_deps, pandas as pd
firehose, s3, records = _setup.ex_vars

# List the objects that have been written to the S3 bucket
objects = s3.list_objects(Bucket='sd-vehicle-data')['Contents']

# Create list for collecting dataframes from read files.
dfs = []

# For every object, load it from S3
for obj in objects:
    data_file = s3.get_object(Bucket='sd-vehicle-data', Key=obj['Key'])

    # Load it into a dataframe, specifying a delimiter and column names
    dfs.append(pd.read_csv(data_file['Body'], 
                           delimiter = " ", 
                           names=["record_id", "timestamp", "vin", "lon", "lat", "speed"]))

# Concatenate the resulting dataframes.
data = pd.concat(dfs)
print(data.groupby(['vin'])['speed'].max())


Membaca data Firehose

Streaming in the cloud

The next step in your streaming data journey is learning how to use transformational lambda functions to go serverless. Through hands-on exercises, you will add lambda layers and trigger lambda functions on specific conditions.

Going serverless

Creating and running Lambda functions

Serverless vs servers

Lambda function components

Reading data from S3 based on an event

Your first live lambda!

Adding a lambda layer

Lambda flow

Using environment variables

Serverless data workflow

Fast vs slow lambdas

A time-triggered lambda

Time-triggered lambda code

Serverless APIs

Create an API lambda

A serverless API

You're now ready to encode and decode streaming data and analyze data directly in the stream. You will even use multiple streams to get daily vehicle top speeds.

A transformational Lambda

Transforming data inside a stream

Encoding and decoding base64

Create a transformational lambda

A barebones transformational lambda

Analyzing data in the stream

Creating a Kinesis data analytics application

Kinesis data analytics vs transformational Lambda

Building a Kinesis data analytics application

Kinesis data analytics SQL components

Using multiple streams

Delivering data from Kinesis Analytics

Get the daily top speed

Using Kinesis data analytics for alerts

Analyzing streaming data

In this final chapter, you will discover how to monitor your stream's performance using logs, metrics, alarms and dashboards. You will use Elasticsearch and build your own Kibana dashboard.

Streaming data case study

Creating an Elasticsearch cluster

RedShift vs Elasticsearch

Monitoring performance

Cloudwatch dasbhoards and alarms

Cloudwatch components

Cloudwatch monitoring flow

Visualizing streaming data

Working with ElasticSearch using Kibana

Cloudwatch vs Elasticsearch

Elasticsearch visualization flow

An alternative approach

Another alternative approach

Monitoring and visualizing streaming data

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.

Latihan izin akses

Streaming Data dengan AWS Kinesis dan Lambda

Latihan interaktif praktis