Apache Pig

-- Load raw sales data
sales_data = LOAD '/data/sales.csv' USING PigStorage(',')
             AS (transaction_id:int, product:chararray, category:chararray, 
                 amount:double, customer_id:int, date:chararray);

-- Filter valid transactions
valid_sales = FILTER sales_data BY amount > 0 AND customer_id IS NOT NULL;

-- Group by category and calculate aggregates
category_group = GROUP valid_sales BY category;
category_stats = FOREACH category_group GENERATE 
                   group AS category,
                   COUNT(valid_sales) AS transaction_count,
                   SUM(valid_sales.amount) AS total_revenue,
                   AVG(valid_sales.amount) AS avg_amount;

-- Sort by revenue and save top categories
sorted_categories = ORDER category_stats BY total_revenue DESC;
top_categories = LIMIT sorted_categories 10;

STORE top_categories INTO '/output/top_categories' USING PigStorage(',');

REGISTER 'myudfs.jar';
DEFINE ExtractIP com.mycompany.pig.ExtractIP();
DEFINE ParseUserAgent com.mycompany.pig.ParseUserAgent();

-- Load web server logs
logs = LOAD '/logs/access.log' AS (line:chararray);

-- Parse log entries
parsed_logs = FOREACH logs GENERATE 
    ExtractIP(line) AS client_ip,
    REGEX_EXTRACT(line, '\[(.*?)\]', 1) AS timestamp,
    REGEX_EXTRACT(line, '"(\w+)\s+(\S+)\s+HTTP', 1) AS method,
    REGEX_EXTRACT(line, '"(\w+)\s+(\S+)\s+HTTP', 2) AS url,
    (int)REGEX_EXTRACT(line, '\s(\d{3})\s', 1) AS status_code,
    ParseUserAgent(line) AS user_agent;

-- Analyze traffic patterns
hourly_traffic = GROUP parsed_logs BY 
    SUBSTRING(timestamp, 0, 13); -- Group by hour

traffic_stats = FOREACH hourly_traffic GENERATE 
    group AS hour,
    COUNT(parsed_logs) AS requests,
    COUNT(FILTER(parsed_logs, status_code >= 400)) AS errors,
    (double)COUNT(FILTER(parsed_logs, status_code >= 400)) / 
    COUNT(parsed_logs) * 100 AS error_rate;

STORE traffic_stats INTO '/output/hourly_traffic' USING PigStorage('\t');