Amazon Athena for Data Engineers

Tables and partitions, partition projection, CTAS/INSERT/UNLOAD, Iceberg tables, cost optimisation, querying from Python, useful SQL, federation and troubleshooting.

1. What Amazon Athena Is

Amazon Athena is serverless SQL over data in S3 (and other sources). It runs on a Trino/Presto-based engine, uses the Glue Data Catalog for table definitions, and charges per terabyte scanned (about $5/TB, 10 MB minimum per query), so data layout directly controls cost.

2. Creating Tables & Partitions

CREATE EXTERNAL TABLE silver.transactions (
  txn_id      string,
  account_id  string,
  channel     string,
  amount      decimal(18,2),
  event_ts    timestamp
)
PARTITIONED BY (dt string)
STORED AS PARQUET
LOCATION 's3://acme-lake-prod/silver/transactions/'
TBLPROPERTIES ('parquet.compression' = 'SNAPPY');

-- register partitions: scan all folders (slow on large tables)…
MSCK REPAIR TABLE silver.transactions;
-- …or add exactly the one you just wrote
ALTER TABLE silver.transactions ADD IF NOT EXISTS PARTITION (dt = '2026-10-26')
LOCATION 's3://acme-lake-prod/silver/transactions/dt=2026-10-26/';

Partition projection: no partition management at all

CREATE EXTERNAL TABLE raw.clickstream (user_id string, url string, ts timestamp)
PARTITIONED BY (dt string)
STORED AS PARQUET
LOCATION 's3://acme-lake-prod/raw/clickstream/'
TBLPROPERTIES (
  'projection.enabled'          = 'true',
  'projection.dt.type'          = 'date',
  'projection.dt.format'        = 'yyyy-MM-dd',
  'projection.dt.range'         = '2024-01-01,NOW',
  'projection.dt.interval'      = '1',
  'projection.dt.interval.unit' = 'DAYS',
  'storage.location.template'   = 's3://acme-lake-prod/raw/clickstream/dt=${dt}/'
);

With projection, Athena computes partition locations from the query filter instead of reading the catalog, so new days are queryable immediately and planning stays fast with thousands of partitions. Queries must filter on the projected column to benefit.

3. CTAS, INSERT INTO & UNLOAD

-- convert raw CSV into partitioned, compressed Parquet in one statement
CREATE TABLE silver.orders_parquet
WITH (
  format = 'PARQUET',
  write_compression = 'SNAPPY',
  external_location = 's3://acme-lake-prod/silver/orders_parquet/',
  partitioned_by = ARRAY['dt']
) AS
SELECT order_id, customer_id, CAST(amount AS decimal(18,2)) AS amount,
       date_format(order_ts, '%Y-%m-%d') AS dt        -- partition column must be last
FROM raw.orders_csv;

-- append a new day
INSERT INTO silver.orders_parquet
SELECT order_id, customer_id, CAST(amount AS decimal(18,2)), date_format(order_ts, '%Y-%m-%d')
FROM raw.orders_csv WHERE date(order_ts) = date '2026-10-26';

-- export results to S3 without creating a table
UNLOAD (SELECT * FROM gold.daily_sales WHERE dt = '2026-10-26')
TO 's3://acme-exports/daily_sales/2026-10-26/'
WITH (format = 'PARQUET', compression = 'SNAPPY');

A single CTAS or INSERT INTO can write at most 100 partitions; for larger backfills, loop over date ranges.

4. Iceberg Tables in Athena

CREATE TABLE silver.customers (
  customer_id string, name string, city string, updated_at timestamp)
PARTITIONED BY (bucket(16, customer_id))
LOCATION 's3://acme-lake-prod/warehouse/silver/customers/'
TBLPROPERTIES ('table_type' = 'ICEBERG');

MERGE INTO silver.customers t
USING staging.customer_updates s ON t.customer_id = s.customer_id
WHEN MATCHED THEN UPDATE SET name = s.name, city = s.city, updated_at = s.updated_at
WHEN NOT MATCHED THEN INSERT (customer_id, name, city, updated_at)
                      VALUES (s.customer_id, s.name, s.city, s.updated_at);

SELECT * FROM silver.customers FOR TIMESTAMP AS OF TIMESTAMP '2026-10-25 00:00:00 UTC';

OPTIMIZE silver.customers REWRITE DATA USING BIN_PACK;   -- compact small files
VACUUM silver.customers;                                  -- expire old snapshots, remove orphans

5. Performance & Cost Optimisation

6. Running Queries from Python

import time, boto3

athena = boto3.client("athena")

def run_query(sql: str, workgroup: str = "data-eng") -> str:
    qid = athena.start_query_execution(QueryString=sql, WorkGroup=workgroup)["QueryExecutionId"]
    while True:
        status = athena.get_query_execution(QueryExecutionId=qid)["QueryExecution"]["Status"]
        if status["State"] in ("SUCCEEDED", "FAILED", "CANCELLED"):
            break
        time.sleep(2)
    if status["State"] != "SUCCEEDED":
        raise RuntimeError(status.get("StateChangeReason", status["State"]))
    return qid

qid = run_query("SELECT channel, count(*) AS txns FROM silver.transactions WHERE dt = '2026-10-26' GROUP BY 1")
for page in athena.get_paginator("get_query_results").paginate(QueryExecutionId=qid):
    for row in page["ResultSet"]["Rows"][1:]:  # first row is the header
        print([c.get("VarCharValue") for c in row["Data"]])

import awswrangler as wr
df = wr.athena.read_sql_query("SELECT * FROM gold.daily_sales WHERE dt = '2026-10-26'",
                              database="gold", workgroup="data-eng")

7. Useful SQL for Lake Data

-- JSON stored as strings
SELECT json_extract_scalar(payload, '$.customer.id') AS customer_id
FROM raw.events;

-- arrays: one row per element
SELECT o.order_id, item.sku, item.qty
FROM silver.orders o
CROSS JOIN UNNEST(o.items) AS t(item);

-- dates and timestamps
SELECT date_trunc('hour', event_ts), date_add('day', -7, current_date),
       from_iso8601_timestamp('2026-10-26T10:15:00Z'), at_timezone(event_ts, 'Asia/Kolkata')
FROM silver.transactions LIMIT 10;

-- which files and partitions are being read?
SELECT "$path", "$file_size", count(*) FROM silver.transactions WHERE dt = '2026-10-26' GROUP BY 1, 2;

8. Federated Queries & Security

• Federated queries use Lambda-based connectors (DynamoDB, RDS/Aurora, Redshift, CloudWatch Logs, others) so one SQL statement can join S3 data with operational stores.
• Lake Formation enforces table, column, row and cell permissions for Athena users, instead of granting raw S3 access.
• Encrypt query results (SSE-KMS) through workgroup settings, and enforce them so users cannot override.
• CloudTrail records every query execution for audit.

9. Common Interview Questions