Kinesis, Firehose, Athena, Glue và họ analytics: clickstream của shop-app thành báo cáo

Ý tưởng tự nhiên, và cũng là sai lầm số một khi dev backend bắt đầu làm analytics: nhét dữ liệu phân tích vào database giao dịch. Bài này đưa clickstream của shop-app đi một đường riêng: app đẩy event → Kinesis → S3 → Athena query SQL. Đó cũng là "bộ ba" mà đề SAA hỏi nhiều nhất trong domain analytics. Phần còn lại của họ analytics (Redshift, EMR, OpenSearch, QuickSight, Lake Formation) chỉ cần nhớ từ khóa, ta để ở bảng cuối.

1. Bài toán: 2 câu hỏi của marketing

Câu hỏi                                            Dữ liệu cần
1. Sản phẩm nào xem nhiều nhất mỗi giờ?             mọi lượt xem (page_view) có productId, timestamp
2. Khách xem rồi bỏ giỏ (add_to_cart không checkout)? chuỗi event theo session: view → add_to_cart → (không) checkout

Quy mô: shop-app ~2,000 request/phút giờ cao điểm → ~1 triệu page_view/ngày, ~300 MB JSON/ngày
Nhu cầu đọc: báo cáo theo giờ/ngày, truy vấn ad-hoc, không cần real-time dưới 1 phút (trừ dashboard "đang hot")

2. Vấn đề của "thêm bảng page_views vào RDS"

• Ghi tốn tài nguyên DB giao dịch: 1 triệu INSERT/ngày cộng index, VACUUM (xem bài MVCC) ăn vào IOPS của shop-db vốn chỉ db.t3.micro; checkout chậm vì analytics.
• Query phân tích quét toàn bảng (GROUP BY product_id, date_trunc('hour', ts)) khóa buffer pool; row-store không hợp với aggregate trên hàng trăm triệu dòng.
• Dung lượng và chi phí: lưu 2 năm clickstream trong RDS gp3 đắt hơn S3 nhiều lần, và backup RDS cũng phình theo.
• Mất event khi DB bận: nếu INSERT đồng bộ trong request và DB chậm, request của user chậm theo; nếu bất đồng bộ không có buffer, event rơi khi app restart.
• Khó mở rộng: mai muốn đưa thêm log ALB, event SQS shop-order-events vào cùng chỗ để join thì RDS không phải nơi làm điều đó.

3. Giải pháp: tách đường "analytics" khỏi "transaction"

OLTP (giao dịch)                  OLAP / analytics
RDS shop-db  ←── checkout         S3 data lake ←── Firehose ←── Kinesis / app ←── page_view event
  row store, index, ACID            Parquet, partition theo ngày/giờ, rẻ, không giới hạn
  query theo khóa, ms              Athena/Redshift quét hàng tỷ dòng, giây → phút

Bốn dịch vụ chính và thuật ngữ đề thi:

Nguyên tắc tiết kiệm Athena mà đề luôn hỏi: Parquet + partition + compression giảm TB scan hàng chục lần → giảm tiền hàng chục lần.

Kinesis Data Streams vs Firehose (phân biệt số 1 trong đề):

Cần real-time (ms), nhiều app cùng đọc, replay, tự viết consumer (KCL, Lambda, Flink)  → Data Streams
Chỉ cần "đổ vào S3/Redshift/OpenSearch" gần real-time, không muốn vận hành, không replay → Firehose
Cần cả hai: Data Streams làm nguồn → Firehose làm một consumer đổ vào S3

Cho shop-app: real-time chưa cần, consumer duy nhất là "đổ vào S3" → app gửi thẳng vào Firehose bằng SDK PutRecord. Nếu sau này cần dashboard "đang hot" real-time hoặc fraud detection, chèn Data Streams vào trước.

4. Cấu hình và code từng bước

Bước 1: Firehose stream shop-clickstream → S3, buffer 5 MB / 60 giây, nén GZIP, partition theo giờ (bước nâng cao: convert Parquet qua Glue schema ở Lab).

aws firehose create-delivery-stream --profile shop-lab \
  --delivery-stream-name shop-clickstream --delivery-stream-type DirectPut \
  --extended-s3-destination-configuration '{
    "RoleARN": "arn:aws:iam::'$ACCOUNT_ID':role/shop-firehose-role",
    "BucketARN": "arn:aws:s3:::shop-app-logs-'$ACCOUNT_ID'",
    "Prefix": "clickstream/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/",
    "ErrorOutputPrefix": "clickstream-errors/!{firehose:error-output-type}/",
    "BufferingHints": { "SizeInMBs": 5, "IntervalInSeconds": 60 },
    "CompressionFormat": "GZIP"
  }' --tags Key=Project,Value=shop-app Key=Env,Value=lab

Prefix dạng year=…/month=… là Hive-style partition: Athena hiểu ngay và chỉ đọc thư mục cần.

Bước 2: Spring Boot gửi event bằng SDK v2. Không gửi đồng bộ trong request; gom vào buffer và đẩy batch bằng PutRecordBatch (tối đa 500 record / 4 MB một lần).

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>firehose</artifactId>
</dependency>

public record PageViewEvent(String eventType, String sessionId, Long userId, long productId,
                            String category, Instant ts, String userAgent) {}

@Component
public class ClickstreamPublisher {
    private static final Logger log = LoggerFactory.getLogger(ClickstreamPublisher.class);
    private final FirehoseClient firehose = FirehoseClient.builder()
            .region(Region.AP_SOUTHEAST_1).build();          // credential: role EC2/ECS qua default chain
    private final ObjectMapper mapper;
    private final BlockingQueue<PageViewEvent> queue = new LinkedBlockingQueue<>(10_000);
    private final String streamName;

    public ClickstreamPublisher(ObjectMapper mapper, @Value("${shop.clickstream.stream}") String streamName) {
        this.mapper = mapper; this.streamName = streamName;
    }

    /** gọi từ controller: không block request */
    public void publish(PageViewEvent e) {
        if (!queue.offer(e)) log.warn("clickstream buffer full, dropping event");   // analytics được phép rơi, checkout thì không
    }

    @Scheduled(fixedDelay = 1000)
    void flush() {
        List<PageViewEvent> batch = new ArrayList<>(500);
        queue.drainTo(batch, 500);
        if (batch.isEmpty()) return;
        List<Record> records = batch.stream()
            .map(e -> Record.builder().data(SdkBytes.fromUtf8String(toJsonLine(e))).build())
            .toList();
        PutRecordBatchResponse res = firehose.putRecordBatch(
            PutRecordBatchRequest.builder().deliveryStreamName(streamName).records(records).build());
        if (res.failedPutCount() > 0) {
            // retry những record lỗi (throttling) – đưa lại queue
            for (int i = 0; i < res.requestResponses().size(); i++)
                if (res.requestResponses().get(i).errorCode() != null) queue.offer(batch.get(i));
            log.warn("firehose failed {} records", res.failedPutCount());
        }
    }

    private String toJsonLine(PageViewEvent e) {
        try { return mapper.writeValueAsString(e) + "\n"; }   // mỗi record một dòng → Athena đọc JSON Lines
        catch (JsonProcessingException ex) { throw new UncheckedIOException(ex); }
    }
}

Ký tự \n cuối record là chi tiết hay quên: Firehose chỉ nối byte lại, không tự thêm xuống dòng; thiếu nó file S3 sẽ là một dòng JSON dính liền và Athena đọc sai.

@GetMapping("/products/{id}")
public ProductDto view(@PathVariable long id, HttpSession session, HttpServletRequest req,
                       @AuthenticationPrincipal(expression = "id") Long userId) {
    ProductDto p = productService.get(id);
    clickstream.publish(new PageViewEvent("page_view", session.getId(), userId, id,
            p.category(), Instant.now(), req.getHeader("User-Agent")));
    return p;
}

Role shop-app-ec2-role cần firehose:PutRecord, firehose:PutRecordBatch trên ARN stream này. Nếu đã dùng Spring Cloud AWS, có thể dùng KinesisTemplate (cho Data Streams); Firehose thì SDK trực tiếp như trên là gọn nhất.

Bước 3: Glue crawler tạo bảng. Crawler quét s3://shop-app-logs-…/clickstream/, suy ra schema JSON, nhận ra partition year/month/day/hour, tạo table clickstream trong database shop_analytics.

aws glue create-database --profile shop-lab --database-input '{"Name":"shop_analytics"}'
aws glue create-crawler --profile shop-lab --name shop-clickstream-crawler \
  --role arn:aws:iam::$ACCOUNT_ID:role/shop-glue-role --database-name shop_analytics \
  --targets '{"S3Targets":[{"Path":"s3://shop-app-logs-'$ACCOUNT_ID'/clickstream/"}]}' \
  --schedule "cron(5 * * * ? *)" \
  --tags Project=shop-app,Env=lab
aws glue start-crawler --profile shop-lab --name shop-clickstream-crawler

Schedule mỗi giờ để thêm partition mới. Cách rẻ hơn và không cần crawler: partition projection trong Athena (khai báo pattern của partition, Athena tự suy ra, không cần metadata).

Bước 4: Athena trả lời 2 câu hỏi của marketing.

-- Câu 1: top sản phẩm theo giờ, chỉ quét partition hôm nay
SELECT hour, productid, count(*) AS views
FROM shop_analytics.clickstream
WHERE year = '2026' AND month = '10' AND day = '21' AND eventtype = 'page_view'
GROUP BY hour, productid
ORDER BY hour, views DESC;

-- Câu 2: session có add_to_cart mà không checkout trong ngày
WITH s AS (
  SELECT sessionid,
         max(CASE WHEN eventtype = 'add_to_cart' THEN 1 ELSE 0 END) AS added,
         max(CASE WHEN eventtype = 'checkout'    THEN 1 ELSE 0 END) AS bought
  FROM shop_analytics.clickstream
  WHERE year = '2026' AND month = '10' AND day = '21'
  GROUP BY sessionid)
SELECT count(*) AS abandoned_sessions, round(100.0 * count(*) / (SELECT count(*) FROM s WHERE added = 1), 1) AS abandon_pct
FROM s WHERE added = 1 AND bought = 0;

Chạy bằng CLI:

QID=$(aws athena start-query-execution --profile shop-lab \
  --query-string "SELECT eventtype, count(*) FROM shop_analytics.clickstream WHERE year='2026' AND month='10' GROUP BY eventtype" \
  --work-group shop-analytics \
  --result-configuration OutputLocation=s3://shop-app-logs-$ACCOUNT_ID/athena-results/ \
  --query QueryExecutionId --output text)
aws athena get-query-results --profile shop-lab --query-execution-id $QID --query 'ResultSet.Rows[].Data[].VarCharValue'
aws athena get-query-execution --profile shop-lab --query-execution-id $QID --query 'QueryExecution.Statistics.DataScannedInBytes'

Bước 5: Chuyển sang Parquet để rẻ hơn 10×. Hai cách: (a) Firehose record format conversion JSON → Parquet dùng schema từ Glue table (bật ở Lab); (b) Athena CTAS định kỳ chuyển dữ liệu cũ:

CREATE TABLE shop_analytics.clickstream_parquet
WITH (format = 'PARQUET', parquet_compression = 'SNAPPY',
      external_location = 's3://shop-app-logs-ACCOUNT/clickstream-parquet/',
      partitioned_by = ARRAY['day'])
AS SELECT eventtype, sessionid, userid, productid, category, ts, useragent,
          concat(year, '-', month, '-', day) AS day
FROM shop_analytics.clickstream;

Cột-store chỉ đọc cột trong SELECT, nén tốt; query "top productId" chỉ đụng 2 cột thay vì cả JSON. Đây là nội dung câu "reduce Athena cost" trong đề.

5. Cách hoạt động

 shop-app (EC2/ECS)                 Amazon Data Firehose                    S3 shop-app-logs-$ACCOUNT_ID
 ┌──────────────────┐   PutRecordBatch  ┌─────────────────────────┐   PUT   ┌───────────────────────────────────┐
 │ controller       │ ────────────────► │ shop-clickstream        │ ──────► │ clickstream/year=2026/month=10/   │
 │  └ publish()     │  ≤500 rec/4 MB    │  buffer 5 MB | 60 s     │ gzip    │   day=21/hour=09/shop-clickstream-│
 │ BlockingQueue    │                   │  (Lambda transform)     │         │   1-2026-10-21-09-00-12-xxxx.gz   │
 │  └ flush() 1 s   │                   │  (→ Parquet qua Glue)   │         │ clickstream-errors/…              │
 └──────────────────┘                   └─────────────────────────┘         └──────────────┬────────────────────┘
                                                                                           │ quét schema, partition
                                                                                           ▼
                                                                            Glue Data Catalog
                                                                            db shop_analytics · table clickstream
                                                                            partitions (year,month,day,hour)
                                                                                           │ metadata
                                                                                           ▼
   Marketing ──SQL──► Athena workgroup shop-analytics ──đọc partition cần──► S3 ──kết quả──► athena-results/
                      (trả tiền theo TB scan)                                                 │
   QuickSight dashboard ◄──────────── SPICE / direct query ───────────────────────────────────┘

Dòng thời gian một event: user xem sản phẩm 09:00:12 → vào queue → 09:00:13 PutRecordBatch → Firehose giữ trong buffer → 09:01:12 (interval 60 s) ghi file .gz vào hour=09/ → 10:05 crawler thêm partition hour=09 (hoặc partition projection khỏi chờ) → query thấy ngay. Độ trễ end-to-end 1–65 phút tùy crawler: đó là nghĩa của "near real-time".

6. So sánh kiểu đề thi

Quy tắc chọn trong đề: ad-hoc + S3 + rẻ + không vận hành → Athena; warehouse, BI nặng, join phức tạp lặp lại → Redshift; phải chạy Spark/Hadoop → EMR; tìm kiếm text/log dashboards → OpenSearch.

Bảng "từ khóa đề → dịch vụ" cho phần còn lại (không cần lab, chỉ cần nhận ra):

7. Khi nào dùng / không dùng

• Dùng Firehose → S3 → Athena khi: cần báo cáo theo giờ/ngày, dữ liệu event/log, team không có người vận hành cluster. Chi phí gần 0 lúc không query. Đây là "stack analytics mặc định" cho startup và cho shop-app.
• Thêm Kinesis Data Streams trước Firehose khi: cần consumer thứ hai real-time (Lambda phát hiện gian lận, Flink tính "đang hot" theo cửa sổ 1 phút), hoặc cần replay.
• Chuyển lên Redshift khi: nhiều analyst query liên tục, join nhiều bảng, dashboard cần giây; Athena bắt đầu tốn hơn vì quét lặp lại cùng dữ liệu. Redshift Serverless là đường ít đau nhất.
• Không dùng Kinesis cho message cần xử lý từng cái một với ack (đơn hàng) → đó là SQS (22.11). Không dùng Firehose làm "queue" vì không replay.
• Không dùng RDS cho clickstream, như đã phân tích. Nhưng dữ liệu đơn hàng trong RDS vẫn cần vào data lake để join với clickstream: dùng DMS CDC sang S3 hoặc Athena Federated Query (22.18 nói về DMS).

Đúng, và đây là chủ đề mà người làm data lake nào cũng va vào trong tháng đầu. Athena (và mọi engine trên S3) phải mở từng object: mỗi file là một request GetObject, list partition là ListObjects. Hàng chục nghìn file 200 KB chậm hơn nhiều so với vài trăm file 128 MB dù tổng byte bằng nhau, vì chi phí là số lần mở file chứ không chỉ byte quét.

1. Kích cỡ file lý tưởng

Mục tiêu: 128 MB – 1 GB mỗi file, Parquet/ORC, Snappy hoặc ZSTD
shop-app: ~300 MB JSON/ngày ≈ 30 MB Parquet/ngày → 1 file/ngày là đủ!
→ partition theo hour là quá nhỏ cho quy mô này; theo day là hợp lý
→ khi lên 100× (30 GB/ngày) thì mới cần hour

Quy tắc chọn partition: mỗi partition nên có ≥ 100 MB dữ liệu và câu query hay dùng phải lọc được trên cột đó. Partition quá mịn làm số file nổ và crawler/metadata chậm; quá thô làm quét thừa.

2. Chỉnh Firehose

• Buffer tối đa 128 MB / 900 giây. Với 300 MB/ngày (~12 MB/giờ), đặt SizeInMBs=128, IntervalInSeconds=900 → ~4 file/giờ, vẫn nhiều file nhỏ nhưng chấp nhận được cho dữ liệu "nóng" hôm nay.
• Bật record format conversion sang Parquet ngay ở Firehose (cần Glue table làm schema): file nhỏ hơn 5–10×, Athena đọc cột.
• Dynamic partitioning nếu muốn partition theo field trong record (ví dụ category), không chỉ theo timestamp.

3. Compaction cho dữ liệu cũ

Dữ liệu hôm nay chấp nhận file nhỏ để tươi; dữ liệu từ hôm qua trở về trước gom lại. Cách rẻ nhất là Athena CTAS/INSERT INTO chạy hàng ngày (qua EventBridge Scheduler → Lambda hoặc Step Functions):

INSERT INTO shop_analytics.clickstream_parquet
SELECT eventtype, sessionid, userid, productid, category, ts, useragent, '2026-10-20' AS day
FROM shop_analytics.clickstream
WHERE year='2026' AND month='10' AND day='20';
-- rồi lifecycle S3 xóa/đưa clickstream/.../day=20/ (JSON thô) sang Glacier sau 7 ngày

Cách "chuẩn hơn" khi dữ liệu lớn: Glue job Spark với bookmark, hoặc table format Apache Iceberg (Athena hỗ trợ) có OPTIMIZE table REWRITE DATA tự compaction và cho UPDATE/DELETE trên S3 (cần cho GDPR "xóa dữ liệu user X").

4. Kiểm tra bằng số

aws athena get-query-execution --query-execution-id $QID \
  --query 'QueryExecution.Statistics.[DataScannedInBytes,TotalExecutionTimeInMillis]'
# So sánh cùng query trên clickstream (JSON gz) và clickstream_parquet: thường thấy byte giảm 10–20×, thời gian giảm 3–5×

Workgroup có thể đặt per-query data scan limit (ví dụ 10 GB) để một câu SELECT * quên WHERE không đốt tiền.

Bạn hiểu đúng rằng chúng cùng một họ "partitioned log" (8.6): Kinesis shard tương ứng Kafka partition, partition key tương ứng message key, sequence number tương ứng offset, KCL tương ứng consumer group. Khác biệt nằm ở ba chỗ: giới hạn cứng, mô hình giá, và API.

Chọn thế nào

Đã có code Kafka, cần Kafka Streams/Connect/exactly-once transaction, giữ dữ liệu dài  → MSK (hoặc MSK Serverless)
Lưu lượng nhỏ–vừa, muốn serverless thật, tích hợp Lambda/Firehose, không có đội Kafka    → Kinesis Data Streams
Chỉ cần đổ vào S3/OpenSearch                                                            → Firehose (không cần cả hai)
Đề thi nói "existing Kafka applications / minimal code change"                          → MSK
Đề thi nói "real-time", "multiple consumers", "shards", "Lambda"                        → Kinesis

Lưu ý kỹ thuật khi viết consumer Kinesis bằng Java: KCL 2.x cần một bảng DynamoDB để lưu checkpoint (giống __consumer_offsets), và mỗi shard chỉ nên có một worker; ProvisionedThroughputExceededException là dấu hiệu hot shard do partition key lệch (ví dụ dùng category làm key khi 70% lượt xem là "điện thoại") – sửa bằng key phân tán hơn (sessionId) hoặc on-demand mode.

Lab 22.17: Clickstream shop-app → Firehose → S3 → Glue → Athena

Mục tiêu

• Tạo Firehose stream shop-clickstream ghi GZIP JSON vào s3://shop-app-logs-$ACCOUNT_ID/clickstream/ với partition Hive-style theo giờ.
• Gửi ≥ 2,000 event từ một chương trình Java (hoặc CLI) và thấy file xuất hiện trong S3 sau ≤ 60 giây.
• Glue crawler tạo table shop_analytics.clickstream với 4 partition key; Athena trả lời "top 5 sản phẩm" và "tỉ lệ bỏ giỏ".
• Chuyển sang Parquet bằng CTAS và đo DataScannedInBytes giảm.

Chi phí & thời gian

• Thời gian: 60–75 phút.
• Ngoài Free Tier: Firehose tính theo GB ingest (khoảng vài cent/GB, lab < 1 MB), Glue crawler tính theo DPU-giờ (tối thiểu 10 phút, khoảng 0.07–0.1 USD/lần chạy), Athena khoảng 5 USD/TB scan (lab quét vài MB ≈ 0). S3 trong Free Tier. Tổng lab < 0.5 USD nếu chạy crawler 2–3 lần.
• Không cần EC2/ALB: gửi event từ máy local bằng profile shop-lab.

Chuẩn bị

• Bucket shop-app-logs-$ACCOUNT_ID (lab 22.6). Java 21 + Maven trên máy local nếu chạy producer Java (có phương án CLI thay thế).

export AWS_REGION=ap-southeast-1
export ACCOUNT_ID=$(aws sts get-caller-identity --profile shop-lab --query Account --output text)
export BUCKET=shop-app-logs-$ACCOUNT_ID

Các bước

Bước 1: Role cho Firehose ghi S3.

cat > fh-trust.json <<EOF
{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"firehose.amazonaws.com"},
 "Action":"sts:AssumeRole","Condition":{"StringEquals":{"sts:ExternalId":"$ACCOUNT_ID"}}}]}
EOF
aws iam create-role --profile shop-lab --role-name shop-firehose-role --assume-role-policy-document file://fh-trust.json \
  --tags Key=Project,Value=shop-app Key=Env,Value=lab
cat > fh-policy.json <<EOF
{"Version":"2012-10-17","Statement":[
 {"Effect":"Allow","Action":["s3:AbortMultipartUpload","s3:GetBucketLocation","s3:GetObject","s3:ListBucket",
   "s3:ListBucketMultipartUploads","s3:PutObject"],
  "Resource":["arn:aws:s3:::$BUCKET","arn:aws:s3:::$BUCKET/*"]},
 {"Effect":"Allow","Action":["glue:GetTable","glue:GetTableVersion","glue:GetTableVersions"],"Resource":"*"},
 {"Effect":"Allow","Action":["logs:PutLogEvents"],"Resource":"arn:aws:logs:$AWS_REGION:$ACCOUNT_ID:log-group:/aws/kinesisfirehose/*"}]}
EOF
aws iam put-role-policy --profile shop-lab --role-name shop-firehose-role --policy-name s3-write --policy-document file://fh-policy.json
sleep 10   # IAM eventual consistency

Console: IAM → Roles → Create role → Custom trust policy. (Console Firehose cũng tự tạo role nếu bạn tạo stream bằng Console.)

Bước 2: Tạo Firehose stream.

aws firehose create-delivery-stream --profile shop-lab \
  --delivery-stream-name shop-clickstream --delivery-stream-type DirectPut \
  --extended-s3-destination-configuration "{
    \"RoleARN\": \"arn:aws:iam::$ACCOUNT_ID:role/shop-firehose-role\",
    \"BucketARN\": \"arn:aws:s3:::$BUCKET\",
    \"Prefix\": \"clickstream/year=!{timestamp:yyyy}/month=!{timestamp:MM}/day=!{timestamp:dd}/hour=!{timestamp:HH}/\",
    \"ErrorOutputPrefix\": \"clickstream-errors/!{firehose:error-output-type}/!{timestamp:yyyy-MM-dd}/\",
    \"BufferingHints\": { \"SizeInMBs\": 1, \"IntervalInSeconds\": 60 },
    \"CompressionFormat\": \"GZIP\",
    \"CloudWatchLoggingOptions\": { \"Enabled\": true, \"LogGroupName\": \"/aws/kinesisfirehose/shop-clickstream\", \"LogStreamName\": \"S3Delivery\" }
  }" --tags Key=Project,Value=shop-app Key=Env,Value=lab
aws logs create-log-group --profile shop-lab --log-group-name /aws/kinesisfirehose/shop-clickstream 2>/dev/null
aws logs create-log-stream --profile shop-lab --log-group-name /aws/kinesisfirehose/shop-clickstream --log-stream-name S3Delivery 2>/dev/null
aws firehose describe-delivery-stream --profile shop-lab --delivery-stream-name shop-clickstream \
  --query 'DeliveryStreamDescription.DeliveryStreamStatus'      # CREATING → ACTIVE (~1 phút)

Console: Amazon Data Firehose → Create Firehose stream → Source: Direct PUT, Destination: S3 → Bucket prefix / Buffer hints / Compression.

Buffer 1 MB / 60 s chỉ để lab thấy kết quả nhanh; prod đặt 64–128 MB / 300–900 s như câu hỏi phụ đã bàn.

Bước 3: Gửi event giả lập. Cách A – Java (project nhỏ clickstream-producer, một file):

// pom: software.amazon.awssdk:firehose (dùng BOM software.amazon.awssdk:bom) + com.fasterxml.jackson.core:jackson-databind
public class ClickstreamProducer {
    record Event(String eventType, String sessionId, Long userId, long productId, String category, String ts, String userAgent) {}
    static final String[] TYPES = {"page_view","page_view","page_view","page_view","add_to_cart","checkout"};
    static final String[] CATS  = {"phone","laptop","audio","accessory"};

    public static void main(String[] args) throws Exception {
        var mapper = new ObjectMapper();
        var rnd = new Random();
        try (var fh = FirehoseClient.builder().region(Region.AP_SOUTHEAST_1)
                .credentialsProvider(ProfileCredentialsProvider.create("shop-lab")).build()) {
            for (int batch = 0; batch < 10; batch++) {
                List<Record> records = new ArrayList<>();
                for (int i = 0; i < 200; i++) {
                    long productId = 1 + (long) Math.floor(Math.abs(rnd.nextGaussian()) * 20) % 100;   // lệch: vài sản phẩm hot
                    var e = new Event(TYPES[rnd.nextInt(TYPES.length)], "s-" + rnd.nextInt(400),
                            rnd.nextBoolean() ? (long) rnd.nextInt(1000) : null, productId,
                            CATS[(int) (productId % CATS.length)], Instant.now().toString(), "lab-producer/1.0");
                    records.add(Record.builder().data(SdkBytes.fromUtf8String(mapper.writeValueAsString(e) + "\n")).build());
                }
                var res = fh.putRecordBatch(b -> b.deliveryStreamName("shop-clickstream").records(records));
                System.out.printf("batch %d sent, failed=%d%n", batch, res.failedPutCount());
            }
        }
    }
}

Cách B – CLI (chậm hơn, đủ cho lab): 300 record lẻ bằng put-record.

for i in $(seq 1 300); do
  T=$([ $((i % 6)) -eq 0 ] && echo add_to_cart || ([ $((i % 13)) -eq 0 ] && echo checkout || echo page_view))
  P=$(( (RANDOM % 7) * (RANDOM % 7) % 50 + 1 ))
  REC="{\"eventType\":\"$T\",\"sessionId\":\"s-$((RANDOM % 80))\",\"userId\":$((RANDOM % 500)),\"productId\":$P,\"category\":\"cat-$((P % 4))\",\"ts\":\"$(date -u +%FT%TZ)\",\"userAgent\":\"cli\"}"
  aws firehose put-record --profile shop-lab --delivery-stream-name shop-clickstream \
    --record "{\"Data\":\"$(printf '%s\n' "$REC" | base64 -w0)\"}" --query RecordId --output text >/dev/null
done; echo done

Lưu ý CLI: Data phải base64 (CLI v2 với --cli-binary-format raw-in-base64-out cũng được), và vẫn phải có \n cuối mỗi record.

Bước 4: Chờ ≤ 60 giây, xem file trong S3.

sleep 70
aws s3 ls s3://$BUCKET/clickstream/ --recursive --profile shop-lab --human-readable | tail -5
# clickstream/year=2026/month=10/day=21/hour=03/shop-clickstream-1-2026-10-21-03-12-45-....gz
aws s3 cp s3://$BUCKET/$(aws s3 ls s3://$BUCKET/clickstream/ --recursive --profile shop-lab | tail -1 | awk '{print $4}') - --profile shop-lab | gunzip | head -3

Console: S3 → bucket → clickstream/ → vào từng cấp year/month/day/hour.

Bước 5: Glue database + crawler.

cat > glue-trust.json <<'EOF'
{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"glue.amazonaws.com"},"Action":"sts:AssumeRole"}]}
EOF
aws iam create-role --profile shop-lab --role-name shop-glue-role --assume-role-policy-document file://glue-trust.json \
  --tags Key=Project,Value=shop-app Key=Env,Value=lab
aws iam attach-role-policy --profile shop-lab --role-name shop-glue-role --policy-arn arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
aws iam put-role-policy --profile shop-lab --role-name shop-glue-role --policy-name s3-read --policy-document "{
 \"Version\":\"2012-10-17\",\"Statement\":[{\"Effect\":\"Allow\",\"Action\":[\"s3:GetObject\",\"s3:ListBucket\"],
 \"Resource\":[\"arn:aws:s3:::$BUCKET\",\"arn:aws:s3:::$BUCKET/*\"]}]}"
sleep 10
aws glue create-database --profile shop-lab --database-input '{"Name":"shop_analytics","Description":"shop-app analytics lab"}'
aws glue create-crawler --profile shop-lab --name shop-clickstream-crawler \
  --role arn:aws:iam::$ACCOUNT_ID:role/shop-glue-role --database-name shop_analytics \
  --targets "{\"S3Targets\":[{\"Path\":\"s3://$BUCKET/clickstream/\"}]}" \
  --configuration '{"Version":1.0,"Grouping":{"TableGroupingPolicy":"CombineCompatibleSchemas"}}' \
  --tags Project=shop-app,Env=lab
aws glue start-crawler --profile shop-lab --name shop-clickstream-crawler
# chờ 1–3 phút
aws glue get-crawler --profile shop-lab --name shop-clickstream-crawler --query 'Crawler.[State,LastCrawl.Status]'
aws glue get-table --profile shop-lab --database-name shop_analytics --name clickstream \
  --query 'Table.[StorageDescriptor.Columns[].Name, PartitionKeys[].Name]'
# [["eventtype","sessionid","userid","productid","category","ts","useragent"],["year","month","day","hour"]]

Console: AWS Glue → Crawlers → Create crawler → Data source S3 path → IAM role → Target database → Run.

Bước 6: Athena workgroup có giới hạn scan, rồi query.

aws athena create-work-group --profile shop-lab --name shop-analytics \
  --configuration "ResultConfiguration={OutputLocation=s3://$BUCKET/athena-results/},EnforceWorkGroupConfiguration=true,BytesScannedCutoffPerQuery=1073741824" \
  --tags Key=Project,Value=shop-app Key=Env,Value=lab

q() {  # helper: chạy query, chờ, in kết quả + byte scan
  local QID=$(aws athena start-query-execution --profile shop-lab --work-group shop-analytics \
    --query-execution-context Database=shop_analytics --query-string "$1" --query QueryExecutionId --output text)
  while [ "$(aws athena get-query-execution --profile shop-lab --query-execution-id $QID --query QueryExecution.Status.State --output text)" = "RUNNING" ]; do sleep 2; done
  aws athena get-query-results --profile shop-lab --query-execution-id $QID --query 'ResultSet.Rows[].Data[].VarCharValue' --output text
  aws athena get-query-execution --profile shop-lab --query-execution-id $QID --query 'QueryExecution.Statistics.[DataScannedInBytes,TotalExecutionTimeInMillis]' --output text
}
q "SELECT eventtype, count(*) c FROM clickstream GROUP BY eventtype ORDER BY c DESC"
q "SELECT productid, count(*) views FROM clickstream WHERE eventtype='page_view' GROUP BY productid ORDER BY views DESC LIMIT 5"
q "WITH s AS (SELECT sessionid, max(CASE WHEN eventtype='add_to_cart' THEN 1 ELSE 0 END) added, max(CASE WHEN eventtype='checkout' THEN 1 ELSE 0 END) bought FROM clickstream GROUP BY sessionid) SELECT count(*) abandoned, round(100.0*count(*)/(SELECT count(*) FROM s WHERE added=1),1) pct FROM s WHERE added=1 AND bought=0"

Console: Athena → Query editor → Workgroup shop-analytics → Database shop_analytics → chạy SQL, tab "Query stats" thấy Data scanned.

Bước 7: CTAS sang Parquet và so sánh.

q "CREATE TABLE clickstream_parquet WITH (format='PARQUET', parquet_compression='SNAPPY', external_location='s3://$BUCKET/clickstream-parquet/', partitioned_by=ARRAY['day']) AS SELECT eventtype, sessionid, userid, productid, category, ts, useragent, concat(year,'-',month,'-',day) AS day FROM clickstream"
q "SELECT productid, count(*) views FROM clickstream_parquet WHERE eventtype='page_view' GROUP BY productid ORDER BY views DESC LIMIT 5"
# So DataScannedInBytes với query ở bước 6: Parquet chỉ đọc 2 cột

Kiểm tra kết quả

# Firehose có nhận và giao
aws cloudwatch get-metric-statistics --profile shop-lab --namespace AWS/Firehose --metric-name IncomingRecords \
  --dimensions Name=DeliveryStreamName,Value=shop-clickstream --statistics Sum --period 300 \
  --start-time $(date -u -d '30 min ago' +%FT%TZ) --end-time $(date -u +%FT%TZ) --query 'Datapoints[].Sum'
aws cloudwatch get-metric-statistics --profile shop-lab --namespace AWS/Firehose --metric-name DeliveryToS3.Success \
  --dimensions Name=DeliveryStreamName,Value=shop-clickstream --statistics Average --period 300 \
  --start-time $(date -u -d '30 min ago' +%FT%TZ) --end-time $(date -u +%FT%TZ) --query 'Datapoints[].Average'   # 1.0
# Không có file lỗi
aws s3 ls s3://$BUCKET/clickstream-errors/ --profile shop-lab    # trống
# Partition trong catalog
aws glue get-partitions --profile shop-lab --database-name shop_analytics --table-name clickstream --query 'Partitions[].Values' --output text

Mong đợi: IncomingRecords ≈ số event đã gửi (2,000 hoặc 300); 3 query Athena trả kết quả trong 2–5 giây; CTAS Parquet quét ít byte hơn rõ rệt.

Thử thêm

• Firehose → Parquet trực tiếp: aws firehose update-destination với DataFormatConversionConfiguration trỏ tới Glue table clickstream (schema), OutputFormatConfiguration.Serializer.ParquetSerDe. Gửi thêm event, thấy file .parquet. Lưu ý buffer tối thiểu khi convert là 64 MB.
• Transform bằng Lambda: gắn Lambda shop-clickstream-enrich (Java 21) thêm field deviceType từ userAgent; Firehose gọi Lambda theo batch, Lambda trả record với result: "Ok" | "Dropped" | "ProcessingFailed".
• Partition projection thay crawler: ALTER TABLE thêm projection.enabled=true, projection.year.type=integer, projection.year.range=2026,2030, …, storage.location.template=s3://…/clickstream/year=${year}/month=${month}/day=${day}/hour=${hour}/; tắt crawler mà vẫn query được partition mới.
• Chèn Kinesis Data Streams: tạo stream shop-clicks on-demand, đổi Firehose source sang KinesisStreamAsSource, producer dùng KinesisClient.putRecords với partition key sessionId; thêm Lambda consumer thứ hai đếm "đang hot" vào DynamoDB shop-carts… hoặc chỉ in log.
• QuickSight: đăng ký bản Standard (có trial), dataset từ Athena clickstream_parquet, vẽ bar chart top sản phẩm. Nhớ hủy đăng ký sau lab vì tính theo user/tháng.

Dọn tài nguyên

aws firehose delete-delivery-stream --profile shop-lab --delivery-stream-name shop-clickstream
aws glue delete-crawler --profile shop-lab --name shop-clickstream-crawler
aws glue delete-table --profile shop-lab --database-name shop_analytics --name clickstream_parquet
aws glue delete-table --profile shop-lab --database-name shop_analytics --name clickstream
aws glue delete-database --profile shop-lab --name shop_analytics
aws athena delete-work-group --profile shop-lab --work-group shop-analytics --recursive-delete-option
aws s3 rm s3://$BUCKET/clickstream/ --recursive --profile shop-lab
aws s3 rm s3://$BUCKET/clickstream-parquet/ --recursive --profile shop-lab
aws s3 rm s3://$BUCKET/athena-results/ --recursive --profile shop-lab
aws s3 rm s3://$BUCKET/clickstream-errors/ --recursive --profile shop-lab
aws logs delete-log-group --profile shop-lab --log-group-name /aws/kinesisfirehose/shop-clickstream
aws iam delete-role-policy --profile shop-lab --role-name shop-firehose-role --policy-name s3-write
aws iam delete-role --profile shop-lab --role-name shop-firehose-role
aws iam delete-role-policy --profile shop-lab --role-name shop-glue-role --policy-name s3-read
aws iam detach-role-policy --profile shop-lab --role-name shop-glue-role --policy-arn arn:aws:iam::aws:policy/service-role/AWSGlueServiceRole
aws iam delete-role --profile shop-lab --role-name shop-glue-role
# Nếu đã thử thêm: xóa Kinesis stream (shard-giờ tính tiền), Lambda, hủy QuickSight subscription
aws kinesis delete-stream --profile shop-lab --stream-name shop-clicks 2>/dev/null

Nhắc: Kinesis Data Streams provisioned tính tiền theo shard-giờ dù không có dữ liệu; QuickSight tính theo user/tháng; crawler đặt schedule sẽ chạy đều và tốn DPU; Redshift/OpenSearch nếu bạn tò mò tạo thử là những dịch vụ đắt theo giờ.

Q1. Bạn thiết kế pipeline clickstream cho ứng dụng Java thế nào? Vì sao không ghi vào database chính?

🇻🇳 Tiếng Việt:

Em tách analytics khỏi database giao dịch: app gom event vào buffer trong bộ nhớ và đẩy batch bằng PutRecordBatch vào Firehose, không gọi đồng bộ trong request và chấp nhận rơi event khi buffer đầy vì analytics không phải checkout. Firehose buffer rồi ghi Parquet nén vào S3 với partition theo ngày, Glue Data Catalog giữ schema, Athena query SQL trả tiền theo byte quét, QuickSight vẽ dashboard. Không ghi vào RDS vì hàng triệu INSERT và query aggregate quét bảng lớn sẽ ăn IOPS, phình backup và làm chậm giao dịch; row-store không hợp OLAP.

🇬🇧 English:

I separate analytics from the transactional database: the app buffers events in memory and ships them in batches with PutRecordBatch to Firehose, never synchronously in the request path, and it accepts dropping events when the buffer is full because analytics is not checkout. Firehose buffers and writes compressed Parquet to S3 partitioned by day, the Glue Data Catalog holds the schema, Athena runs SQL billed per byte scanned, and QuickSight builds dashboards. I avoid RDS because millions of inserts and large aggregate scans would consume IOPS, bloat backups and slow transactions; a row store is the wrong shape for OLAP.

Q2. Kinesis Data Streams và Firehose khác nhau thế nào? Khi nào dùng cái nào?

🇻🇳 Tiếng Việt:

Data Streams là stream thật: dữ liệu giữ 24 giờ đến 365 ngày, nhiều consumer đọc độc lập và đọc lại được, scale bằng shard hoặc on-demand, latency vài trăm mili giây, mình tự viết consumer bằng KCL, Lambda hay Flink. Firehose là dịch vụ giao hàng: nhận record, buffer theo MB hoặc giây, tùy chọn transform bằng Lambda và chuyển sang Parquet, rồi ghi vào S3, Redshift, OpenSearch hay HTTP endpoint; không lưu, không replay, không shard, near real-time từ 60 giây. Cần real-time và nhiều consumer thì Data Streams; chỉ cần đổ vào kho thì Firehose; cần cả hai thì Data Streams làm nguồn cho Firehose.

🇬🇧 English:

Data Streams is a true stream: data is retained from 24 hours up to 365 days, multiple consumers read independently and can replay, it scales by shards or on-demand, latency is a few hundred milliseconds, and you write consumers with KCL, Lambda or Flink. Firehose is a delivery service: it receives records, buffers by size or time, optionally transforms with Lambda and converts to Parquet, then writes to S3, Redshift, OpenSearch or an HTTP endpoint; no storage, no replay, no shards, near real-time from 60 seconds. For real-time and multiple consumers use Data Streams; to land data in a store use Firehose; for both, Data Streams feeds Firehose.

Q3. Làm sao giảm chi phí và thời gian query Athena?

🇻🇳 Tiếng Việt:

Athena tính theo byte quét nên mọi cách đều nhằm đọc ít byte hơn: dùng định dạng cột Parquet hoặc ORC để chỉ đọc cột trong SELECT, nén Snappy hoặc ZSTD, partition theo cột hay lọc như ngày và luôn lọc trên partition trong WHERE, gom file nhỏ thành file 128 MB đến 1 GB, dùng partition projection thay crawler cho partition theo thời gian, và đặt giới hạn byte quét trong workgroup để chặn câu SELECT * quên WHERE. Với query lặp lại nhiều, cân nhắc CTAS ra bảng tổng hợp hoặc chuyển sang Redshift.

🇬🇧 English:

Athena bills per byte scanned, so everything aims at reading fewer bytes: columnar Parquet or ORC so only selected columns are read, Snappy or ZSTD compression, partitioning by commonly filtered columns such as date and always filtering on partitions in WHERE, compacting small files into 128 MB to 1 GB files, partition projection instead of crawlers for time-based partitions, and a per-query byte limit in the workgroup to stop a SELECT * without WHERE. For heavily repeated queries, pre-aggregate with CTAS or move to Redshift.

Q4. Khi nào chọn Redshift thay Athena, và EMR thay Glue?

🇻🇳 Tiếng Việt:

Athena cho truy vấn ad-hoc, không vận hành, dữ liệu để yên trên S3, trả theo query. Redshift khi có nhiều analyst query liên tục, join phức tạp nhiều bảng, cần giây ổn định và BI nặng; Redshift Spectrum vẫn query được S3 cho dữ liệu lạnh, Serverless giảm vận hành. Glue cho ETL serverless bằng Spark khi mình không cần kiểm soát cluster; EMR khi cần Hadoop ecosystem đầy đủ, thư viện riêng, tuning cluster, hoặc job rất lớn chạy trên Spot để rẻ, với core node giữ HDFS và task node dùng Spot.

🇬🇧 English:

Athena is for ad-hoc queries with no operations, data staying in S3, paid per query. Redshift fits many analysts querying continuously, complex multi-table joins, consistent sub-second to second latency and heavy BI; Spectrum still queries S3 for cold data and Serverless cuts operations. Glue is serverless Spark ETL when you do not need cluster control; EMR is for the full Hadoop ecosystem, custom libraries, cluster tuning or very large jobs running cheaply on Spot, with core nodes holding HDFS and task nodes on Spot.

Góc thi SAA-C03

Cách nhớ ngắn gọn

• OLTP (RDS) tách khỏi OLAP (S3 lake). Event đi: app → (Kinesis) → Firehose → S3 Parquet → Glue Catalog → Athena → QuickSight.
• Streams = real-time, shard, replay, nhiều consumer. Firehose = near real-time, không shard, không lưu, đổ vào đích.
• Athena rẻ khi: Parquet + partition + nén + file to + lọc trên partition. Workgroup giới hạn byte.
• Redshift = warehouse; EMR = Spark/Hadoop tự kiểm soát; Glue = ETL serverless + catalog; OpenSearch = search/log; Lake Formation = quyền cho data lake; Flink = SQL/Java real-time trên stream; MSK = Kafka.