AT&T Marketing Analytics Platform

Executive Summary

Business Challenge

AT&T's marketing operations team faced critical inefficiencies in their data infrastructure and reporting processes:

• Manual Data Extraction: Marketing analysts spent 15-20 hours weekly manually exporting data from Workfront, Bynder, and Webphone into Excel spreadsheets for campaign reporting
• Data Silos: No single source of truth for marketing performance - project data in Workfront, creative assets in Bynder, call tracking in Webphone, with no cross-system visibility
• Reporting Lag: Campaign performance reports took 5-7 days to compile, preventing real-time optimization and budget reallocation
• Data Quality Issues: 27% of reports contained errors from manual data entry, copy-paste mistakes, and formula errors in Excel workbooks
• Limited Analytics: No ability to perform advanced analytics like campaign attribution modeling, asset performance correlation, or multi-touch lead scoring
• Scalability Constraints: Excel-based workflows couldn't handle growing data volumes (500K+ marketing events monthly across 200+ campaigns)
• Stakeholder Frustration: Marketing leadership lacked visibility into campaign ROI, asset utilization rates, and resource allocation efficiency

Solution Architecture

Implementation Process

import requests
import pandas as pd
import boto3
from datetime import datetime, timedelta
import logging

class WebphoneExtractor:
    """
    Extract call tracking data from Webphone API and load to S3.
    Handles pagination, rate limiting, and error recovery.
    """
    
    def __init__(self, api_key: str, api_secret: str):
        self.base_url = "https://api.webphone.com/v2"
        self.session = requests.Session()
        self.session.headers.update({
            'Authorization': f'Bearer {self._get_auth_token(api_key, api_secret)}',
            'Content-Type': 'application/json'
        })
        self.s3_client = boto3.client('s3')
        
    def extract_calls(self, start_date: str, end_date: str) -> pd.DataFrame:
        """Extract call data for date range with pagination"""
        all_calls = []
        page = 1
        page_size = 1000
        
        while True:
            try:
                response = self.session.get(
                    f"{self.base_url}/calls",
                    params={
                        'start_date': start_date,
                        'end_date': end_date,
                        'page': page,
                        'page_size': page_size,
                        'include': 'campaign,lead,disposition'
                    },
                    timeout=30
                )
                
                if response.status_code == 429:  # Rate limit
                    logging.warning("Rate limit hit, waiting 60 seconds...")
                    time.sleep(60)
                    continue
                    
                response.raise_for_status()
                data = response.json()
                
                calls = data.get('data', [])
                if not calls:
                    break
                    
                all_calls.extend(calls)
                logging.info(f"Extracted page {page}: {len(calls)} calls")
                
                # Check if more pages
                if len(calls) < page_size:
                    break
                    
                page += 1
                time.sleep(0.6)  # Rate limiting: 100 req/min = 1 req every 0.6s
                
            except requests.exceptions.RequestException as e:
                logging.error(f"API request failed: {e}")
                if page > 1:  # Partial success, return what we have
                    break
                raise
        
        df = pd.DataFrame(all_calls)
        return self._transform_calls(df)
    
    def _transform_calls(self, df: pd.DataFrame) -> pd.DataFrame:
        """Transform and clean call data"""
        # Parse timestamps
        df['call_datetime'] = pd.to_datetime(df['call_timestamp'])
        df['call_date'] = df['call_datetime'].dt.date
        
        # Extract nested fields
        df['campaign_id'] = df['campaign'].apply(lambda x: x.get('id') if x else None)
        df['campaign_name'] = df['campaign'].apply(lambda x: x.get('name') if x else None)
        df['lead_id'] = df['lead'].apply(lambda x: x.get('id') if x else None)
        df['disposition'] = df['disposition'].apply(lambda x: x.get('name') if x else 'Unknown')
        
        # Calculate metrics
        df['call_duration_seconds'] = pd.to_numeric(df['duration'], errors='coerce')
        df['is_qualified_lead'] = df['disposition'].isin(['Qualified', 'Meeting Scheduled', 'Demo Booked'])
        df['is_conversion'] = df['disposition'].isin(['Sale Closed', 'Contract Signed'])
        
        # Select relevant columns
        columns = [
            'call_id', 'call_date', 'call_datetime', 'phone_number',
            'campaign_id', 'campaign_name', 'lead_id', 'disposition',
            'call_duration_seconds', 'is_qualified_lead', 'is_conversion',
            'caller_city', 'caller_state', 'tracking_number'
        ]
        
        return df[columns]
    
    def upload_to_s3(self, df: pd.DataFrame, bucket: str, prefix: str):
        """Upload extracted data to S3 as parquet"""
        if df.empty:
            logging.warning("No data to upload")
            return
            
        # Partition by date for efficient Snowpipe ingestion
        for date, group_df in df.groupby('call_date'):
            file_name = f"{prefix}/year={date.year}/month={date.month:02d}/day={date.day:02d}/calls_{datetime.now().strftime('%Y%m%d_%H%M%S')}.parquet"
            
            # Convert to parquet
            parquet_buffer = io.BytesIO()
            group_df.to_parquet(parquet_buffer, engine='pyarrow', compression='snappy')
            parquet_buffer.seek(0)
            
            # Upload to S3
            self.s3_client.put_object(
                Bucket=bucket,
                Key=file_name,
                Body=parquet_buffer.getvalue()
            )
            
            logging.info(f"Uploaded {len(group_df)} records to s3://{bucket}/{file_name}")
    
    def run_daily_extract(self, bucket: str, prefix: str = 'webphone/calls'):
        """Main extraction workflow for daily scheduled run"""
        # Extract previous day's data
        yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
        
        logging.info(f"Starting Webphone extraction for {yesterday}")
        
        df = self.extract_calls(start_date=yesterday, end_date=yesterday)
        
        logging.info(f"Extracted {len(df)} calls, {df['is_qualified_lead'].sum()} qualified leads")
        
        self.upload_to_s3(df, bucket, prefix)
        
        return {
            'date': yesterday,
            'total_calls': len(df),
            'qualified_leads': int(df['is_qualified_lead'].sum()),
            'conversions': int(df['is_conversion'].sum())
        }


# AWS Lambda handler
def lambda_handler(event, context):
    """Lambda function for scheduled daily extraction"""
    import os
    
    # Initialize extractor
    extractor = WebphoneExtractor(
        api_key=os.environ['WEBPHONE_API_KEY'],
        api_secret=os.environ['WEBPHONE_API_SECRET']
    )
    
    # Run extraction
    result = extractor.run_daily_extract(
        bucket=os.environ['S3_BUCKET'],
        prefix='webphone/calls'
    )
    
    logging.info(f"Extraction complete: {result}")
    
    return {
        'statusCode': 200,
        'body': json.dumps(result)
    }

-- models/marts/marketing/fct_campaign_attribution.sql
{{
    config(
        materialized='incremental',
        unique_key='attribution_id',
        cluster_by=['attribution_date', 'campaign_id'],
        tags=['marketing', 'attribution']
    )
}}

WITH calls AS (
    SELECT
        call_id,
        call_date AS attribution_date,
        campaign_id,
        lead_id,
        is_qualified_lead,
        is_conversion,
        call_duration_seconds
    FROM {{ ref('stg_webphone__calls') }}
    {% if is_incremental() %}
        WHERE call_date > (SELECT MAX(attribution_date) FROM {{ this }})
    {% endif %}
),

campaigns AS (
    SELECT
        campaign_id,
        campaign_name,
        campaign_type,
        channel,
        start_date,
        end_date,
        budget_amount,
        campaign_owner
    FROM {{ ref('stg_workfront__campaigns') }}
),

assets AS (
    SELECT
        asset_id,
        campaign_id,
        asset_type,
        download_count,
        view_count
    FROM {{ ref('int_bynder__campaign_assets') }}
),

-- Multi-touch attribution using linear model
lead_touchpoints AS (
    SELECT
        lead_id,
        COUNT(DISTINCT campaign_id) AS touchpoint_count,
        ARRAY_AGG(DISTINCT campaign_id) AS campaign_journey,
        MIN(call_date) AS first_touch_date,
        MAX(call_date) AS last_touch_date,
        DATEDIFF('day', MIN(call_date), MAX(call_date)) AS journey_length_days
    FROM calls
    GROUP BY lead_id
),

-- Revenue allocation (equal credit to all touchpoints)
attributed_revenue AS (
    SELECT
        c.lead_id,
        c.campaign_id,
        c.call_date,
        CASE 
            WHEN c.is_conversion THEN 
                3500 / lt.touchpoint_count  -- Avg deal size $3,500
            ELSE 0
        END AS attributed_revenue,
        1.0 / lt.touchpoint_count AS attribution_weight
    FROM calls c
    INNER JOIN lead_touchpoints lt ON c.lead_id = lt.lead_id
)

SELECT
    {{ dbt_utils.generate_surrogate_key(['c.call_id', 'c.campaign_id']) }} AS attribution_id,
    c.call_date AS attribution_date,
    c.campaign_id,
    camp.campaign_name,
    camp.channel,
    c.lead_id,
    c.is_qualified_lead,
    c.is_conversion,
    lt.touchpoint_count,
    lt.journey_length_days,
    ar.attribution_weight,
    ar.attributed_revenue,
    
    -- Campaign metrics
    camp.budget_amount AS campaign_budget,
    COALESCE(a.asset_count, 0) AS assets_used,
    COALESCE(a.total_downloads, 0) AS asset_downloads,
    
    -- Calculated metrics
    CASE 
        WHEN camp.budget_amount > 0 THEN 
            ar.attributed_revenue / camp.budget_amount
        ELSE NULL 
    END AS campaign_roi,
    
    -- Metadata
    CURRENT_TIMESTAMP() AS _loaded_at
    
FROM calls c
INNER JOIN campaigns camp ON c.campaign_id = camp.campaign_id
LEFT JOIN lead_touchpoints lt ON c.lead_id = lt.lead_id
LEFT JOIN attributed_revenue ar ON c.call_id = ar.call_id AND c.campaign_id = ar.campaign_id
LEFT JOIN (
    SELECT 
        campaign_id,
        COUNT(DISTINCT asset_id) AS asset_count,
        SUM(download_count) AS total_downloads
    FROM assets
    GROUP BY campaign_id
) a ON c.campaign_id = a.campaign_id

{% if is_incremental() %}
WHERE c.call_date > (SELECT MAX(attribution_date) FROM {{ this }})
{% endif %}

Results & Business Impact

Technical Challenges & Solutions

Key Learnings & Best Practices

Technologies & Tools