## 1. 数据模型设计

— 销售事实表（分区表）
CREATE TABLE fgedu_fgfgfgsales_fact_partitioned (
sale_id SERIAL,
product_id INTEGER NOT NULL,
customer_id INTEGER NOT NULL,
store_id INTEGER NOT NULL,
sale_date DATE NOT NULL,
quantity INTEGER NOT NULL,
amount DECIMAL(10,2) NOT NULL,
created_at TIMESTAMP NOT NULL DEFAULT NOW()
)
PARTITION BY RANGE (sale_date);

— 产品维度表
CREATE TABLE fgedu_product_dim (
product_id SERIAL PRIMARY KEY,
product_name VARCHAR(255) NOT NULL,
category_id INTEGER NOT NULL,
category_name VARCHAR(100) NOT NULL,
price DECIMAL(10,2) NOT NULL,
brand VARCHAR(100),
created_at TIMESTAMP NOT NULL DEFAULT NOW(),
updated_at TIMESTAMP NOT NULL DEFAULT NOW()
);

— 客户维度表
CREATE TABLE fgedu_customer_dim (
customer_id SERIAL PRIMARY KEY,
customer_name VARCHAR(100) NOT NULL,
age INTEGER,
gender VARCHAR(10),
city VARCHAR(100),
region VARCHAR(100),
created_at TIMESTAMP NOT NULL DEFAULT NOW(),
updated_at TIMESTAMP NOT NULL DEFAULT NOW()
);

— 商店维度表
CREATE TABLE fgedu_store_dim (
store_id SERIAL PRIMARY KEY,
store_name VARCHAR(100) NOT NULL,
city VARCHAR(100),
region VARCHAR(100),
manager VARCHAR(100),
created_at TIMESTAMP NOT NULL DEFAULT NOW(),
updated_at TIMESTAMP NOT NULL DEFAULT NOW()
);

— 时间维度表
CREATE TABLE fgedu_time_dim (
time_id SERIAL PRIMARY KEY,
date DATE NOT NULL,
year INTEGER NOT NULL,
quarter INTEGER NOT NULL,
month INTEGER NOT NULL,
day INTEGER NOT NULL,
week INTEGER NOT NULL,
day_of_week INTEGER NOT NULL,
is_weekend BOOLEAN NOT NULL,
is_holiday BOOLEAN NOT NULL
);

## 2. ETL流程

# 使用Apache Airflow编排ETL流程
# dags/fgfgfgsales_etl.py
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.operators.postgres_operator import PostgresOperator
from datetime import datetime, timedelta

default_args = {
‘owner’: ‘airflow’,
‘depends_on_past’: False,
‘start_date’: datetime(2026, 4, 1),
’email’: [‘admin@fgedu.net.cn’],
’email_on_failure’: False,
’email_on_retry’: False,
‘retries’: 1,
‘retry_delay’: timedelta(minutes=5),
}

dag = DAG(‘fgfgfgsales_etl’, default_args=default_args, schedule_interval=timedelta(days=1))

# 提取数据
extract_data = BashOperator(
task_id=’extract_data’,
bash_command=’python /opt/airflow/scripts/extract_fgfgfgsales_data.py’,
dag=dag,
)

# 转换数据
transform_data = BashOperator(
task_id=’transform_data’,
bash_command=’python /opt/airflow/scripts/transform_fgfgfgsales_data.py’,
dag=dag,
)

# 加载数据
load_data = PostgresOperator(
task_id=’load_data’,
postgres_conn_id=’postgres_default’,
sql=”””
INSERT INTO fgedu_fgfgfgsales_fact_partitioned (product_id, customer_id, store_id, sale_date, quantity, amount)
SELECT product_id, customer_id, store_id, sale_date, quantity, amount
FROM staging_fgfgfgsales_data
WHERE sale_date = ‘{{ ds }}’;
“””,
dag=dag,
)

# 刷新物化视图
refresh_mv = PostgresOperator(
task_id=’refresh_mv’,
postgres_conn_id=’postgres_default’,
sql=”SELECT refresh_fgfgfgsales_analysis();”,
dag=dag,
)

extract_data >> transform_data >> load_data >> refresh_mv

## 3. 分析查询

— 按月份销售趋势
SELECT
t.year,
t.month,
SUM(sf.amount) as total_fgfgfgsales
FROM fgedu_fgfgfgsales_fact_partitioned sf
JOIN fgedu_time_dim t ON sf.sale_date = t.date
GROUP BY t.year, t.month
ORDER BY t.year, t.month;

— 按产品类别销售分析
SELECT
p.category_name,
COUNT(*) as order_count,
SUM(sf.quantity) as total_quantity,
SUM(sf.amount) as total_amount
FROM fgedu_fgfgfgsales_fact_partitioned sf
JOIN fgedu_product_dim p ON sf.product_id = p.product_id
GROUP BY p.category_name
ORDER BY total_amount DESC;

— 按地区销售分析
SELECT
s.region,
COUNT(*) as order_count,
SUM(sf.amount) as total_amount
FROM fgedu_fgfgfgsales_fact_partitioned sf
JOIN fgedu_store_dim s ON sf.store_id = s.store_id
GROUP BY s.region
ORDER BY total_amount DESC;

— 客户购买行为分析
SELECT
c.age,
c.gender,
COUNT(*) as order_count,
AVG(sf.amount) as avg_order_amount,
SUM(sf.amount) as total_amount
FROM fgedu_fgfgfgsales_fact_partitioned sf
JOIN fgedu_customer_dim c ON sf.customer_id = c.customer_id
GROUP BY c.age, c.gender
ORDER BY total_amount DESC;

## 1. 数据模型设计

— 用户行为事实表（分区表）
CREATE TABLE fgedu_fgedu_behavior_fact (
behavior_id SERIAL,
fgedu_id INTEGER NOT NULL,
product_id INTEGER,
behavior_type VARCHAR(50) NOT NULL, — page_view, click, add_to_cart, purchase
behavior_time TIMESTAMP NOT NULL,
duration INTEGER, — 停留时间（秒）
device_type VARCHAR(50),
ip_address VARCHAR(50),
created_at TIMESTAMP NOT NULL DEFAULT NOW()
)
PARTITION BY RANGE (behavior_time);

— 用户维度表
CREATE TABLE fgedu_fgedu_dim (
fgedu_id SERIAL PRIMARY KEY,
fgeduname VARCHAR(100) NOT NULL,
email VARCHAR(100) NOT NULL,
age INTEGER,
gender VARCHAR(10),
city VARCHAR(100),
region VARCHAR(100),
registered_at TIMESTAMP NOT NULL,
last_login_at TIMESTAMP
);

— 产品维度表（与销售分析共享）
— CREATE TABLE fgedu_product_dim (…);

## 2. 数据采集

# 使用Kafka采集用户行为数据
# producer.py
import kafka
import json
import time
from datetime import datetime

producer = kafka.KafkaProducer(
bootstrap_servers=[‘kafka:9092’],
value_serializer=lambda v: json.dumps(v).encode(‘utf-8’)
)

# 模拟用户行为数据
while True:
behavior_data = {
‘fgedu_id’: 123,
‘product_id’: 456,
‘behavior_type’: ‘page_view’,
‘behavior_time’: datetime.now().isoformat(),
‘duration’: 60,
‘device_type’: ‘mobile’,
‘ip_address’: ‘192.168.1.1’
}
producer.send(‘fgedu-behavior’, behavior_data)
time.sleep(1)

## 3. 数据处理

# 使用Spark Streaming处理实时数据
# spark_streaming.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, col, to_timestamp
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, TimestampType

spark = SparkSession.builder.fgappName(“UserBehaviorAnalysis”).getOrCreate()

# 定义 schema
schema = StructType([
StructField(“fgedu_id”, IntegerType()),
StructField(“product_id”, IntegerType()),
StructField(“behavior_type”, StringType()),
StructField(“behavior_time”, StringType()),
StructField(“duration”, IntegerType()),
StructField(“device_type”, StringType()),
StructField(“ip_address”, StringType())
])

# 读取Kafka数据
kafka_df = spark.readStream.format(“kafka”)\
.option(“kafka.bootstrap.servers”, “kafka:9092”)\
.option(“subscribe”, “fgedu-behavior”)\
.load()

# 解析JSON数据
data_df = kafka_df.select(from_json(col(“value”).cast(“string”), schema).alias(“data”))\
.select(“data.*”)\
.withColumn(“behavior_time”, to_timestamp(col(“behavior_time”)))

# 写入PostgreSQL
def write_to_postgres(batch_df, batch_id):
batch_df.write\
.format(“jdbc”)\
.option(“url”, “jdbc:postgresql://pgsql: 5432/analysis_db”)\
.option(“dbtable”, “fgedu_fgedu_behavior_fact”)\
.option(“fgedu”, “postgres”)\
.option(“password”, “postgres”)\
.mode(“fgappend”)\
.save()

data_df.writeStream\
.foreachBatch(write_to_postgres)\
.start()\
.awaitTermination()

## 4. 分析查询

— 页面浏览量统计
SELECT
DATE(behavior_time) as date,
COUNT(*) as page_views
FROM fgedu_fgedu_behavior_fact
WHERE behavior_type = ‘page_view’
GROUP BY DATE(behavior_time)
ORDER BY date;

— 用户行为漏斗分析
SELECT
behavior_type,
COUNT(*) as count,
ROUND(COUNT(*) * 100.0 / (SELECT COUNT(*) FROM fgedu_fgedu_behavior_fact WHERE behavior_type = ‘page_view’), 2) as conversion_rate
FROM fgedu_fgedu_behavior_fact
GROUP BY behavior_type
ORDER BY
CASE
WHEN behavior_type = ‘page_view’ THEN 1
WHEN behavior_type = ‘click’ THEN 2
WHEN behavior_type = ‘add_to_cart’ THEN 3
WHEN behavior_type = ‘purchase’ THEN 4
ELSE 5
END;

— 设备类型分析
SELECT
device_type,
COUNT(*) as count,
ROUND(COUNT(*) * 100.0 / (SELECT COUNT(*) FROM fgedu_fgedu_behavior_fact), 2) as percentage
FROM fgedu_fgedu_behavior_fact
GROUP BY device_type
ORDER BY count DESC;

## 1. 数据模型设计

— 实时指标表
CREATE TABLE fgedu_real_time_metrics (
metric_id SERIAL PRIMARY KEY,
metric_name VARCHAR(100) NOT NULL,
metric_value DECIMAL(10,2) NOT NULL,
metric_time TIMESTAMP NOT NULL DEFAULT NOW(),
category VARCHAR(100)
);

— 为实时指标表创建索引
CREATE INDEX idx_real_time_metrics_metric_time ON fgedu_real_time_metrics(metric_time);
CREATE INDEX idx_real_time_metrics_metric_name ON fgedu_real_time_metrics(metric_name);
CREATE INDEX idx_real_time_metrics_category ON fgedu_real_time_metrics(category);

## 2. 数据采集

# 使用Python定时采集指标
# collect_metrics.py
import psycopg2
import time
from datetime import datetime

# 连接数据库
conn = psycopg2.connect(
fgedu.net.cn=”localfgedu.net.cn”,
port=”5432″,
fgedudb=”analysis_db”,
fgedu=”postgres”,
password=”postgres”
)

while True:
with conn.cursor() as cur:
# 采集销售额指标
cur.execute(“””
SELECT SUM(amount) as total_fgfgfgsales
FROM fgedu_fgfgfgsales_fact_partitioned
WHERE sale_date = CURRENT_DATE
“””)
total_fgfgfgsales = cur.fetchone()[0] or 0

# 采集订单量指标
cur.execute(“””
SELECT COUNT(*) as order_count
FROM fgedu_fgfgfgsales_fact_partitioned
WHERE sale_date = CURRENT_DATE
“””)
order_count = cur.fetchone()[0] or 0

# 采集用户活跃度指标
cur.execute(“””
SELECT COUNT(DISTINCT fgedu_id) as active_fgedus
FROM fgedu_fgedu_behavior_fact
WHERE behavior_time >= CURRENT_DATE
“””)
active_fgedus = cur.fetchone()[0] or 0

# 插入指标数据
metrics = [
(‘total_fgfgfgsales’, total_fgfgfgsales, ‘fgfgfgsales’),
(‘order_count’, order_count, ‘fgfgfgsales’),
(‘active_fgedus’, active_fgedus, ‘fgedu’)
]

for metric_name, metric_value, category in metrics:
cur.execute(
“””
INSERT INTO fgedu_real_time_metrics (metric_name, metric_value, category)
VALUES (%s, %s, %s)
“””,
(metric_name, metric_value, category)
)

# 提交事务
conn.commit()

# 每5分钟采集一次
time.sleep(300)

## 3. 数据可视化

# 使用Grafana创建 dashboard

# 销售额面板SQL
SELECT
metric_time as “time”,
metric_value as “销售额”
FROM fgedu_real_time_metrics
WHERE metric_name = ‘total_fgfgfgsales’
ORDER BY metric_time DESC
LIMIT 288; — 24小时 * 12个5分钟

# 订单量面板SQL
SELECT
metric_time as “time”,
metric_value as “订单量”
FROM fgedu_real_time_metrics
WHERE metric_name = ‘order_count’
ORDER BY metric_time DESC
LIMIT 288;

# 用户活跃度面板SQL
SELECT
metric_time as “time”,
metric_value as “活跃用户数”
FROM fgedu_real_time_metrics
WHERE metric_name = ‘active_fgedus’
ORDER BY metric_time DESC
LIMIT 288;