Data & Analytics

Data & Analytics

A dashboard says yesterday's revenue fell by 40%. Did customers buy less, did a delivery pipeline fall behind, or did a new schema turn prices into nulls? A useful analytics system lets you answer that question before acting on the number.

This lesson follows an online shop's events from ingestion to a governed report. You will build four applications: a validated event delivery path, a lake-to-warehouse report, a streaming alert, and a restricted analyst view. At each stage, define what counts as complete, how duplicates are handled, and how the result can be checked. Prices and dataset sizes used in arithmetic are invented teaching inputs, not current AWS quotes.

1. Give the data a contract

For this shop, one accepted event has a nonempty event_id, a timezone-aware event_time, a customer_id, a two-letter region_code, and an integer amount_cents from 0 through 9,223,372,036,854,775,807, matching the nonnegative portion of the curated BIGINT type. Amounts are in one agreed currency; do not add cents from different currencies. An event ID identifies one business event, including after a retry. A correction needs a documented revision policy rather than silently reusing an ID with different content.

Record both business time and ingestion time. A purchase at 23:59 UTC can arrive after midnight. Partitioning by arrival date and filtering by purchase date are different operations. A trusted ingestion gateway authenticates the producer and stamps source_id, schema_version=1 and ingestion_time before publishing the event. Ingestion time here means gateway acceptance time, not Firehose delivery time. Preserve these values on a retry and carry them through normalized and curated storage. Both timestamps must fall from 2000-01-01 inclusive to 2100-01-01 exclusive in UTC, this shop’s supported date range. Keep the source identity and schema version so a report can be traced back to its inputs.

Separate raw, rejected and curated data. These are processing roles, not permissions: a prefix called private/ provides no protection by itself. Preserve raw input according to the retention policy, quarantine malformed records, and publish only validated datasets. Reconcile counts and monetary totals at every boundary. Missing data is not a zero-dollar sale.

Gateway -> Kinesis Data Streams -> Firehose transform
                                         |       |
                                      accepted  rejected
                                         |       |
                                         v       v
                                    S3 landing  restricted error prefix
                                         |
                           validate + deduplicate + compact
                                         |
                                         v
                               S3 curated Parquet
                                         |
                         Glue Data Catalog: schema/location
                              /                       \
                   Athena queries               Redshift/Spectrum
                              \                       /
                               authorized, checked reports

The catalog stores metadata; it is not the files themselves. A crawler discovers schema and partitions, but successful discovery does not establish that an amount is valid or that all expected events arrived. Schema evolution needs review: adding an optional field differs from changing cents into dollars. Glue Data Catalog

2. Choose the service for the work

Amazon Redshift is an analytical warehouse, with provisioned and serverless deployment options. Columnar storage can reduce work when a query needs a small subset of columns. In a provisioned cluster, the leader plans work, compute nodes execute it, and the leader assembles results for the client. Redshift Spectrum can query external data in S3; it does not require copying every lake dataset into local warehouse tables. Redshift architecture, Spectrum

Amazon Athena supports SQL analytics, including queries over S3 data described in the catalog. Engine version 3 incorporates work from Trino and Presto; treat Athena's documented engine behavior as the contract rather than assuming an arbitrary upstream version. Athena has different pricing modes, including SQL scan-based pricing and provisioned capacity, alongside Spark offerings. A bytes-scanned calculation applies to the selected scan-based SQL model, not every Athena workload. Engine version 3, Athena pricing

AWS Glue supplies data integration, including managed Spark ETL and visual job authoring. A transform such as ApplyMapping changes fields and types; it does not prove every source value was converted correctly. Inspect conversion errors, required columns and reconciliation results. For a bookmark-enabled job, initialize with job.init, use stable source transformation_ctx values, and call job.commit after successful work. Bookmarks track supported source processing; replay-safe outputs remain your responsibility. Resetting a bookmark does not clean the destination. Glue transforms, Bookmark use, Bookmark details

Kinesis Data Streams retains events for consumers; Amazon Data Firehose buffers and delivers streaming data to configured destinations. Amazon Managed Service for Apache Flink supports stateful stream processing. The older Kinesis Data Analytics for SQL Applications service was discontinued, with deletion of applications beginning January 27, 2026. Do not start a new design with its SQL pumps. Firehose overview, Flink overview, SQL service discontinuation

3. Application one: deliver valid events without hiding failures

Firehose invokes a transformation Lambda with base64-encoded records. Return one result for each input record, preserving recordId and providing base64 data. Ok, Dropped and ProcessingFailed have different meanings. A malformed purchase is a processing failure in our contract; reporting it as intentionally dropped would conceal data loss. Failed transformations have a documented S3 error path that needs monitoring and appropriately restricted access. Transformation response, Failure handling

This complete transformation handler accepts Firehose's event envelope. It emits newline-delimited JSON for accepted records, keeps only the agreed fields, and returns the original encoded data for rejected records. It deliberately does not deduplicate: invocation-local memory cannot remember every prior batch. Limit incoming record and batch sizes to the supported transformation limits before deploying.

import base64
import binascii
import json
import re
from datetime import datetime, timezone

def parse_shop_time(raw_time):
    if not isinstance(raw_time, str):
        raise ValueError("missing timestamp")
    try:
        instant = datetime.fromisoformat(raw_time.replace("Z", "+00:00"))
        if instant.tzinfo is None or instant.utcoffset() is None:
            raise ValueError("timestamp needs a timezone")
        instant = instant.astimezone(timezone.utc)
    except OverflowError as error:
        raise ValueError("timestamp conversion out of range") from error
    if not (datetime(2000, 1, 1, tzinfo=timezone.utc) <= instant
            < datetime(2100, 1, 1, tzinfo=timezone.utc)):
        raise ValueError("timestamp outside shop contract")
    return instant

def normalize_event(value):
    if not isinstance(value, dict):
        raise ValueError("event must be an object")
    for key in ("event_id", "customer_id", "source_id"):
        if not isinstance(value.get(key), str) or not value[key].strip():
            raise ValueError("missing identifier")
    if type(value.get("schema_version")) is not int or value["schema_version"] != 1:
        raise ValueError("unsupported schema version")
    amount = value.get("amount_cents")
    if type(amount) is not int or not 0 <= amount <= 2**63 - 1:
        raise ValueError("amount must be nonnegative BIGINT cents")
    region = value.get("region_code")
    if not isinstance(region, str) or not re.fullmatch(r"[A-Z]{2}", region):
        raise ValueError("invalid region code")
    instant = parse_shop_time(value.get("event_time"))
    ingestion = parse_shop_time(value.get("ingestion_time"))
    return {"event_id": value["event_id"],
            "customer_id": value["customer_id"],
            "source_id": value["source_id"], "schema_version": 1,
            "ingestion_time": ingestion.isoformat(),
            "event_time": instant.isoformat(),
            "event_date": instant.date().isoformat(),
            "region_code": region, "amount_cents": amount}

def handler(event, context):
    results = []
    for record in event["records"]:
        result = {"recordId": record["recordId"],
                  "result": "ProcessingFailed", "data": record["data"]}
        try:
            raw = base64.b64decode(record["data"], validate=True)
            clean = normalize_event(json.loads(raw))
            line = json.dumps(clean, allow_nan=False) + "\n"
            result.update(result="Ok", data=base64.b64encode(
                line.encode("utf-8")).decode("ascii"))
        except (ValueError, TypeError, UnicodeError, binascii.Error):
            pass
        results.append(result)
    return {"records": results}

The outer envelope keys are supplied by Firehose; the payload provenance fields come from the trusted gateway. This validator checks their shape, while gateway authentication and write permissions establish their trust. The handler does not swallow a programming error or an invalid outer envelope as a successful batch. Test a good purchase, invalid JSON, an empty ID, a negative amount, a boolean amount and a timestamp without a timezone. Check that true is rejected as money even though Python treats booleans as integers in many contexts.

Configure transformation under ExtendedS3DestinationConfiguration.ProcessingConfiguration when creating the corresponding S3 delivery stream. Supply the existing destination bucket, delivery role and Lambda ARN, plus permissions for S3, invocation and any KMS keys. A plain S3DestinationConfiguration is not the container for processing configuration. Begin with a literal landing prefix; writing an event_date field does not automatically create event-date directories. The curated ETL stage must explicitly partition its output. CreateDeliveryStream request

A successful transformation is not proof that S3 delivery or downstream compaction completed. Follow delivery errors and freshness, then reconcile raw, rejected, duplicate and published counts. Firehose buffering also means a near-real-time path is not an instantaneous one.

4. Application two: turn the lake into a trustworthy report

The curated job parses the normalized event time, enforces the schema, applies the event-ID revision policy and writes Parquet with Snappy compression into physical event_date=YYYY-MM-DD/ directories. Compact small files to sizes suitable for the workload. The following Athena DDL assumes those files already exist, use the stated types, and omit the partition field from the file schema. event_date comes from the directory partition. Replace the reviewed bucket name; creating this table does not convert JSON to Parquet or move files.

CREATE EXTERNAL TABLE shop_events (
  event_id string,
  customer_id string,
  source_id string,
  schema_version int,
  ingestion_time string,
  event_time string,
  region_code string,
  amount_cents bigint
)
PARTITIONED BY (event_date string)
STORED AS PARQUET
LOCATION 's3://reviewed-shop-lake/curated/events/';

ALTER TABLE shop_events ADD IF NOT EXISTS
PARTITION (event_date='2026-09-14')
LOCATION 's3://reviewed-shop-lake/curated/events/event_date=2026-09-14/';

SELECT region_code, count(*) AS accepted_events,
       sum(amount_cents) AS revenue_cents
FROM shop_events
WHERE event_date = '2026-09-14'
GROUP BY region_code
ORDER BY region_code;

Register each required partition through a controlled catalog process, or deliberately configure partition projection. A partition predicate enables pruning; declaring a partition column alone does not make every query selective. The report counts curated business events, so repeated delivery must have been resolved before publication. The job must retain the three provenance fields and reject incompatible schema versions. Size aggregate types for the reporting population as well: individually valid BIGINT amounts can still overflow a BIGINT sum. Athena data types With accepted records NO/1250, NO/750 and DE/500, the result is DE: 1 event/500 cents and NO: 2 events/2000 cents. A duplicate NO/1250 would inflate NO revenue to 3250 cents unless handled. Athena partitioning, Columnar formats

For an illustrative scan price of $5 per decimal TB, measured scans of 10 GB cost $0.05 and 1 GB cost $0.005 before billing rounding/minimums or other charges. The calculation is bytes / 10^12 × 5. That particular measured change saves 90%; neither Parquet nor partitioning guarantees it for all queries. Result reuse, selected columns, file layout and predicates affect scans. Do not multiply overlapping savings percentages without measuring the combined query.

This read-only helper checks a previously submitted Athena query. It returns a scan estimate only after SUCCEEDED; it does not interpret a query ID or a running state as a completed report. Supply an authorized client and the query's ID.

from decimal import Decimal

def query_scan_evidence(athena, query_id):
    q = athena.get_query_execution(QueryExecutionId=query_id)["QueryExecution"]
    state = q["Status"]["State"]
    if state != "SUCCEEDED":
        return {"state": state,
                "reason": q["Status"].get("StateChangeReason", "")}
    scanned = q["Statistics"].get("DataScannedInBytes")
    if scanned is None:
        raise ValueError("completed query has no scan evidence")
    return {"state": state, "scanned_bytes": scanned,
            "teaching_scan_cost": Decimal(scanned) / Decimal(10**12) * 5}

For a completed report, retrieve all GetQueryResults pages and validate the result schema, not only the first page. Workgroup scan controls and output-location controls help constrain operation; bytes-based limits are not an all-service spending cap. Query execution API, Query result pagination

When the report belongs in Redshift

Recurring joins and concurrent warehouse workloads may justify loading curated data into Redshift. Distribution choices are AUTO, KEY, ALL and EVEN; the default is AUTO. KEY can colocate matching join keys when both tables and the join align, but one very frequent customer can create skew. ALL replicates a table and adds storage/load work. Evaluate representative plans and distribution, rather than making every ID a distribution key. CREATE TABLE distribution

A materialized view stores computed results. It can reduce repeated aggregation work, but the report must tolerate its refresh behavior and freshness. Do not assume every query shape supports incremental refresh or automatic rewriting. Redshift also performs automatic maintenance; manual VACUUM or ANALYZE should respond to observed needs rather than being a ritual on every read. Materialized views, Vacuum management

For a database migration, AWS DMS full load plus change data capture is a data movement plan, not a complete application/schema migration. Check source logging, connectivity, types and target prerequisites. Reconcile counts and business totals, observe CDC lag, stop or fence writes for the chosen cutover procedure, and validate the target before switching readers. Creating a replication task is not starting it; task status alone is not proof of equivalent business results. DMS Redshift target, Starting replication

5. Application three: alert on a stream with explicit time rules

Suppose each purchase needs to reach both an archive and a rolling regional alert. A stream allows independent consumers and replay within its retention window. Kinesis Data Streams defaults to 24 hours of retention and can be configured up to 365 days. Retention is not a guarantee that a stalled consumer catches up before records expire. Monitor iterator age and recovery throughput. Retention

For a provisioned stream, use both the sustained per-shard write limits of 1 MB/second and 1,000 records/second when estimating shards. Ten thousand 100-byte records per second need at least ten shards by the record-rate calculation even though payload bytes alone are around 1 MB/second. Protocol/accounting details, hot partition keys and headroom can require more. A lower bound is not a safe capacity plan. Shared GetRecords reads have their own per-shard limits; enhanced fan-out changes consumer capacity behavior. Stream quotas

The default maximum record size is 1 MiB, and a stream can be configured for records up to 10 MiB. Large records use intermittent burst capacity; the larger maximum does not turn a shard into a sustained 10 MiB/second writer. Check producer, transformation and consumer limits together before enabling larger records. Large records

A partition key maps records to a shard. It does not establish business ordering across concurrent producers and retries. PutRecords can partly succeed, and retrying failed entries can change their order relative to successful entries. Track per-entry outcomes, retry with bounded backoff, and make consumers deduplicate stable event IDs. Use a documented per-entity ordering/version strategy when later events depend on earlier ones. PutRecords behavior

For Flink, define event timestamps, watermarks, window boundaries and a late-event policy. A watermark expresses event-time progress used by the operator; it is not proof that no late record can ever arrive. Our teaching alert uses a prior completed window as its baseline, excludes the current reading from that baseline, and treats missing or constant history explicitly. Flink time and watermarks

import math
import statistics

def anomaly_evidence(current, prior):
    values = [float(x) for x in prior]
    current = float(current)
    if not math.isfinite(current) or any(not math.isfinite(x) for x in values):
        raise ValueError("measurements must be finite")
    if len(values) < 2:
        return {"status": "insufficient_history"}
    mean = statistics.mean(values)
    sd = statistics.stdev(values)
    if sd == 0:
        return {"status": "constant_baseline", "changed": current != mean}
    z = (current - mean) / sd
    return {"status": "scored", "z": z, "candidate_alert": abs(z) > 2}

For history [8, 10, 12], the sample mean is 10 and sample standard deviation is 2. A current value of 16 has z=3 and becomes a candidate alert. This rule is a teaching detector, not proof of fraud or a guaranteed statistical false-positive rate. Seasonality, correlated observations and a poorly chosen window can make it noisy. Test late arrivals, duplicates, restart from a checkpoint and sink retries in the actual stream job. An offline function does not implement Flink state, watermarks or delivery semantics.

6. Application four: publish a restricted analyst view

An analyst should see only EU rows and selected columns from a governed customer table. Lake Formation can express row and column filters for supported integrated engines. The following request-building function specifies one filter for an existing table with region_group, customer_id and total_cents. It returns the two complete API argument objects; an authorized administrator can review them before calling the corresponding operations. catalog_id, database, table and principal must refer to the reviewed deployment.

def analyst_filter_requests(catalog_id, database, table, principal_arn):
    filter_name = "eu-sales-view"
    create = {"TableData": {
        "TableCatalogId": catalog_id, "DatabaseName": database,
        "TableName": table, "Name": filter_name,
        "RowFilter": {"FilterExpression": "region_group = 'EU'"},
        "ColumnNames": ["customer_id", "total_cents"]}}
    grant = {"Principal": {"DataLakePrincipalIdentifier": principal_arn},
             "Resource": {"DataCellsFilter": {
                 "TableCatalogId": catalog_id, "DatabaseName": database,
                 "TableName": table, "Name": filter_name}},
             "Permissions": ["SELECT"]}
    return create, grant

Pass create to create_data_cells_filter and grant to grant_permissions. Choose ColumnNames or ColumnWildcard as appropriate, not both. Creating a filter alone grants no access. The example is for a same-account governed query path; cross-account sharing adds its own configuration and recipient permissions. Data filtering, CreateDataCellsFilter, GrantPermissions

Check effective permissions, including broader table grants, IAMAllowedPrincipals, hybrid-mode opt-in and direct S3/KMS access. A narrow filtered grant does not subtract permissions from an existing broader grant. An analyst who can directly read raw objects may bypass the governed query path. Restrict query-result locations too: saved results can contain sensitive rows. Lake Formation permissions, Hybrid access

Test as the actual analyst principal. EU rows and allowed columns should be readable; non-EU rows should be absent, and a query requesting a disallowed column should fail. Test direct object access and saved-query-result access separately. After a permission change, repeat both positive and negative tests. A test executed as the administrator does not establish the analyst's restrictions.

7. Practice the whole investigation

Return to the missing revenue. Compare expected source events with raw arrivals, transformation failures, duplicate/revision decisions and curated output. Check the event-date boundary, currency and units. Verify the query finished, all result pages were consumed, and the reporting principal sees the intended population. Only then investigate whether customer behavior changed.

For your own pipeline, write down one acceptance check and one failure-recovery check at each arrow in the diagram. A useful completion criterion is concrete: “all accepted events through the agreed watermark are represented once under our revision policy, rejected events are accounted for, and the report reconciles to those inputs.” That criterion is more informative than “the Glue job was green.”