用 DuckDB 构建营销数据产品框架：从免费工具到规模化变现

-- 自动推断格式，直接读取 Google Ads CSV
CREATE TABLE google_ads AS
READ_CSV('https://ads.google.com/export/data.csv', 
         HEADER TRUE, AUTO DETECT TRUE, 
         LIMIT 1000000);

-- 读取本地或多个 CSV 文件
CREATE TABLE facebook_ads AS
READ_CSV('ad_campaigns/*.csv', 
         HEADER TRUE, AUTO DETECT TRUE);

-- Facebook 广告 API 返回的是嵌套 JSON，DuckDB 可以直接查询
CREATE TABLE facebook_ads_nested AS
SELECT 
    campaign_id,
    json_extract(data, '$.spend') as spend,
    json_extract(data, '$.clicks') as clicks,
    json_extract(data, '$.conversions') as conversions,
    json_extract_array(data, '$.daily_stats') as daily_stats
FROM 'facebook_export.json';

-- 连接 PostgreSQL CRM 数据库
CREATE TABLE crm_leads AS
SELECT customer_id, signup_date, plan_type, lifetime_value
FROM postgresql://user:***@crm.example.com:5432/crm_db.leads;

-- 连接 MySQL 电商订单库
CREATE TABLE orders AS
SELECT order_id, user_id, order_amount, created_at
FROM mysql://user:***@mysql.example.com:3306/ecommerce.orders;

-- 创建统一的市场营销活动事实表
WITH unified_ads AS (
    SELECT 
        campaign_id,
        'google' as platform,
        date,
        impressions, clicks, conversions, spend
    FROM google_ads
    
    UNION ALL
    
    SELECT 
        campaign_id,
        'facebook' as platform,
        date,
        impressions, clicks, conversions, spend
    FROM facebook_ads
),
unified_orders AS (
    SELECT 
        user_id,
        order_id,
        order_amount,
        DATE(created_at) as order_date
    FROM orders
)
-- 最终的分析视图
CREATE VIEW marketing_performance AS
SELECT 
    u.platform,
    u.date,
    SUM(u.impressions) as total_impressions,
    SUM(u.clicks) as total_clicks,
    SUM(u.conversions) as total_conversions,
    SUM(u.spend) as total_spend,
    -- 计算关键指标
    1.0 * SUM(u.clicks)/SUM(u.impressions) as ctr,
    1.0 * SUM(u.conversions)/SUM(u.clicks) as cvr,
    1.0 * SUM(u.spend)/SUM(u.conversions) as cpv,
    SUM(o.order_amount) as revenue,
    SUM(u.spend) as cost,
    SUM(o.order_amount) - SUM(u.spend) as profit,
    (SUM(o.order_amount) - SUM(u.spend)) / NULLIF(SUM(u.spend), 0) as roas
FROM unified_ads u
LEFT JOIN unified_orders o ON u.user_id = o.user_id AND DATE(o.order_date) = u.date
WHERE u.date BETWEEN DATEADD('day', -90, CURRENT_DATE) AND CURRENT_DATE
GROUP BY u.platform, u.date;

-- 创建日维度聚合物化视图
CREATE MATERIALIZED VIEW daily_marketing_summary AS
SELECT 
    DATE(date) as event_day,
    platform,
    SUM(impressions) as daily_impressions,
    SUM(clicks) as daily_clicks,
    SUM(conversions) as daily_conversions,
    SUM(spend) as daily_cost,
    SUM(CASE WHEN status = 'converted' THEN 1 ELSE 0 END) as daily_conversion_count,
    AVG(roas) as avg_roas
FROM marketing_performance
GROUP BY DATE(date), platform;

-- 每晚定时刷新
REFRESH MATERIALIZED VIEW CONCURRENTLY daily_marketing_summary;

-- 创建设备索引加速查询
CREATE INDEX idx_daily_summary_date ON daily_marketing_summary(event_day);
CREATE INDEX idx_daily_summary_platform ON daily_marketing_summary(platform);

# marketing_pipeline.py - 营销数据分析自动化脚本
import duckdb
import pandas as pd
from datetime import datetime, timedelta
import smtplib
from email.mime.text import MIMEText

# 1. 连接 DuckDB
con = duckdb.connect()

# 2. 加载外部数据
print("正在拉取多源数据...")
con.execute(
    "CREATE TEMP TABLE google_ads AS READ_CSV('https://ads.google.com/export/data.csv', HEADER TRUE)"
)

# 3. 执行复杂分析
print("正在运行分析模型...")
result = con.execute(
    """WITH ad_data AS (SELECT * FROM google_ads),
    agg AS (
        SELECT 
            campaign_id,
            SUM(impressions) as impressions,
            SUM(clicks) as clicks,
            SUM(conversions) as conversions,
            SUM(cost) as cost,
            COUNT(DISTINCT date) as days_running
        FROM ad_data
        GROUP BY campaign_id
    )
    SELECT *,
           1.0*clicks/impressions as ctr,
           1.0*conversions/clicks as cvr,
           cost/conversions as cpcv
    FROM agg
    WHERE days_running >= 7
    ORDER BY cpcv ASC"""
).fetchdf()

# 4. 生成可视化
result.to_csv('/tmp/daily_report.csv')

# 5. 发送邮件通知
# send_email(result)

print("报表生成完成！")