Summary  
This chapter covers using SQL to aggregate event-based data by grouping events, counting distinct users at each stage, and ordering them to measure progression through a predefined sequence.

General domain of usage  
marketing analytics

Funnels are a foundational concept in **marketing analytics**, representing the journey users take through a predefined sequence of steps toward a desired action. Understanding **funnels** allows you to pinpoint where users drop off and where they convert, helping you optimize marketing strategies and improve conversion rates. **Event-based data**, which logs user actions as discrete events, is essential for funnel analysis because it enables you to track each user's progression through the funnel in detail.

-- Example funnel_steps table defining the order of funnel events
SELECT *
FROM funnel_steps
ORDER BY step_order;

To analyze a funnel, you need to know how many users reach each step. SQL makes this possible by allowing you to count users at each funnel stage using event data. Typically, you group events by their type and count the number of **distinct users** who performed each event. This approach helps you visualize where users drop off and how many complete the funnel.

-- SQL query to count distinct users at each funnel step
SELECT e.event_type,
  COUNT(DISTINCT e.user_id) AS user_count
FROM events e
GROUP BY e.event_type
ORDER BY MIN(e.event_timestamp);

import unittest
import psycopg2
import pandas as pd

class TestPostgresQuery(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.connection = psycopg2.connect(
            host="localhost",
            user="agent",
            password="qwerty",
            dbname="testdb",
            sslmode="disable"
        )
        cls.cursor = cls.connection.cursor()
        # Seed the events table with data matching expected user counts
        cls.cursor.execute("DROP TABLE IF EXISTS events;")
        cls.cursor.execute("""
            CREATE TABLE events (
                event_id SERIAL PRIMARY KEY,
                user_id INTEGER NOT NULL,
                event_type VARCHAR(50) NOT NULL,
                event_timestamp TIMESTAMP NOT NULL
            );
        """)
        # Insert test data
        # 5 users sign up
        # 3 users view_product
        # 3 users add_to_cart
        # 2 users purchase
        cls.cursor.execute("""
            INSERT INTO events (user_id, event_type, event_timestamp) VALUES
            (1, 'signup', '2023-01-01 08:00:00'),
            (2, 'signup', '2023-01-01 08:01:00'),
            (3, 'signup', '2023-01-01 08:02:00'),
            (4, 'signup', '2023-01-01 08:03:00'),
            (5, 'signup', '2023-01-01 08:04:00'),
            (1, 'view_product', '2023-01-01 08:10:00'),
            (2, 'view_product', '2023-01-01 08:11:00'),
            (3, 'view_product', '2023-01-01 08:12:00'),
            (1, 'add_to_cart', '2023-01-01 08:20:00'),
            (2, 'add_to_cart', '2023-01-01 08:21:00'),
            (3, 'add_to_cart', '2023-01-01 08:22:00'),
            (1, 'purchase', '2023-01-01 08:30:00'),
            (2, 'purchase', '2023-01-01 08:31:00')
        """)
        cls.connection.commit()

    def test_distinct_user_count_per_step(self):
        """Check correct user counts for each funnel step."""
        with open("query.sql", "r") as file:
            query = file.read()

        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)

        expected = {
            'signup': 5,
            'view_product': 3,
            'add_to_cart': 3,
            'purchase': 2
        }
        missing = []
        for step, count in expected.items():
            found = result_df[(result_df['event_type'] == step) & (result_df['user_count'] == count)]
            if found.empty:
                missing.append(f"{step}: {count}")
        condition = len(missing) == 0
        _dynamic_test(
            self,
            condition,
            "User counts for each funnel step are correct.",
            f"Incorrect user counts for steps: {', '.join(missing)}"
        )

    def test_ordering_by_event_timestamp(self):
        """Check that results are ordered by earliest event_timestamp."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        rows = self.cursor.fetchall()
        columns = [desc[0] for desc in self.cursor.description]
        result_df = pd.DataFrame(rows, columns=columns)
        # Get the earliest timestamp for each event_type
        self.cursor.execute("""
            SELECT event_type, MIN(event_timestamp) as min_time
            FROM events
            GROUP BY event_type
            ORDER BY min_time
        """)
        expected_order = [row[0] for row in self.cursor.fetchall()]
        actual_order = result_df['event_type'].tolist()
        condition = expected_order == actual_order
        _dynamic_test(
            self,
            condition,
            "Results are ordered by earliest event_timestamp.",
            f"Results are not ordered by earliest event_timestamp.\\nExpected: {expected_order}\\nActual: {actual_order}"
        )

    def test_required_columns_present(self):
        """Check that event_type and user_count columns are present."""
        with open("query.sql", "r") as file:
            query = file.read()
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        required = {'event_type', 'user_count'}
        condition = required.issubset(set(columns))
        _dynamic_test(
            self,
            condition,
            "Both required columns are present.",
            f"Missing columns: {required - set(columns)}"
        )

def _dynamic_test(test_case, condition, success_message, failure_message):
    if condition:
        test_case._testMethodName = success_message
        test_case.assertTrue(True, success_message)
    else:
        test_case._testMethodName = failure_message
        test_case.fail(failure_message)

def normalize_text(text):
    text = text.lower()
    text = re.sub(r"\\s{2,}", " ", text)
    text = re.sub(r"\\s*([,:?])\\s*", r"\\1 ", text)
    return text.strip()

if __name__ == "__main__":
    unittest.main()


test_solution.py

Master SQL techniques tailored for marketing analytics. Learn to model event-based data, analyze funnels, perform cohort analysis, implement attribution logic, and evaluate campaign performance using real-world marketing datasets.

Learn how to structure and query event-based data for marketing analytics, enabling flexible and powerful insights into user behavior.

Learn to build and analyze marketing funnels using SQL, tracking user progression through key conversion steps.

Master cohort analysis using SQL DATE functions to uncover retention and behavioral trends in marketing data.

Explore attribution logic and campaign performance analysis in SQL to measure marketing effectiveness.

Building Funnels with SQL

解答