# 创建销售数据模型
$ podman exec -it fgedu-postgres psql -U fgedu -d fgedudb

# 创建维度表
CREATE TABLE dim_product (
product_id SERIAL PRIMARY KEY,
product_name VARCHAR(255) NOT NULL,
product_category VARCHAR(100) NOT NULL,
product_price DECIMAL(10, 2) NOT NULL
);

CREATE TABLE dim_customer (
customer_id SERIAL PRIMARY KEY,
customer_name VARCHAR(255) NOT NULL,
customer_address VARCHAR(255) NOT NULL,
customer_email VARCHAR(255) NOT NULL
);

CREATE TABLE dim_date (
date_id SERIAL PRIMARY KEY,
date DATE NOT NULL,
year INT NOT NULL,
month INT NOT NULL,
day INT NOT NULL,
quarter INT NOT NULL
);

# 创建事实表
CREATE TABLE fact_sales (
sales_id SERIAL PRIMARY KEY,
product_id INT REFERENCES dim_product(product_id),
customer_id INT REFERENCES dim_customer(customer_id),
date_id INT REFERENCES dim_date(date_id),
quantity INT NOT NULL,
amount DECIMAL(10, 2) NOT NULL
);

# 插入测试数据
INSERT INTO dim_product (product_name, product_category, product_price) VALUES
(‘Product A’, ‘Category 1’, 100.00),
(‘Product B’, ‘Category 1’, 200.00),
(‘Product C’, ‘Category 2’, 300.00);

INSERT INTO dim_customer (customer_name, customer_address, customer_email) VALUES
(‘Customer 1’, ‘Address 1’, ‘customer1@example.com’),
(‘Customer 2’, ‘Address 2’, ‘customer2@example.com’);

INSERT INTO dim_date (date, year, month, day, quarter) VALUES
(‘2026-01-01’, 2026, 1, 1, 1),
(‘2026-01-02’, 2026, 1, 2, 1),
(‘2026-02-01’, 2026, 2, 1, 1);

INSERT INTO fact_sales (product_id, customer_id, date_id, quantity, amount) VALUES
(1, 1, 1, 2, 200.00),
(2, 1, 1, 1, 200.00),
(3, 2, 2, 1, 300.00),
(1, 2, 3, 3, 300.00);

# 提交事务
COMMIT;

# 退出PostgreSQL
\q

# 创建DAG文件
$ cat > /Podman/fgdata/airflow/dags/sales_etl.py << EOF from airflow import DAG from airflow.operators.postgres_operator import PostgresOperator from datetime import datetime, timedelta default_args = { 'owner': 'fgedu', 'depends_on_past': False, 'start_date': datetime(2026, 4, 10), 'email_on_failure': False, 'email_on_retry': False, 'retries': 1, 'retry_delay': timedelta(minutes=5), } dag = DAG( 'sales_etl', default_args=default_args, description='Sales data ETL workflow', schedule_interval=timedelta(days=1), ) # 提取数据 extract_data = PostgresOperator( task_id='extract_data', postgres_conn_id='postgres_default', sql=""" SELECT p.product_id, p.product_name, p.product_category, p.product_price, c.customer_id, c.customer_name, c.customer_address, c.customer_email, d.date_id, d.date, d.year, d.month, d.day, d.quarter, s.quantity, s.amount FROM fact_sales s JOIN dim_product p ON s.product_id = p.product_id JOIN dim_customer c ON s.customer_id = c.customer_id JOIN dim_date d ON s.date_id = d.date_id """, dag=dag, ) # 转换数据 transform_data = PostgresOperator( task_id='transform_data', postgres_conn_id='postgres_default', sql=""" CREATE TABLE IF NOT EXISTS sales_summary AS SELECT d.year, d.month, p.product_category, SUM(s.quantity) AS total_quantity, SUM(s.amount) AS total_amount FROM fact_sales s JOIN dim_product p ON s.product_id = p.product_id JOIN dim_date d ON s.date_id = d.date_id GROUP BY d.year, d.month, p.product_category """, dag=dag, ) # 加载数据 load_data = PostgresOperator( task_id='load_data', postgres_conn_id='postgres_default', sql=""" INSERT INTO sales_summary (year, month, product_category, total_quantity, total_amount) SELECT d.year, d.month, p.product_category, SUM(s.quantity) AS total_quantity, SUM(s.amount) AS total_amount FROM fact_sales s JOIN dim_product p ON s.product_id = p.product_id JOIN dim_date d ON s.date_id = d.date_id GROUP BY d.year, d.month, p.product_category ON CONFLICT (year, month, product_category) DO UPDATE SET total_quantity = EXCLUDED.total_quantity, total_amount = EXCLUDED.total_amount """, dag=dag, ) extract_data >> transform_data >> load_data
EOF

# 查看DAG文件
$ ls -la /Podman/fgdata/airflow/dags/

# 输出日志
total 12
drwxr-xr-x 2 root root 4096 Apr 10 10:00 .
drwxr-xr-x 5 root root 4096 Apr 10 10:00 ..
-rw-r–r– 1 root root 2048 Apr 10 10:00 sales_etl.py

# 连接PostgreSQL
$ podman exec -it fgedu-postgres psql -U fgedu -d fgedudb

# 查询销售汇总数据
SELECT
d.year,
d.month,
p.product_category,
SUM(s.quantity) AS total_quantity,
SUM(s.amount) AS total_amount
FROM
fact_sales s
JOIN
dim_product p ON s.product_id = p.product_id
JOIN
dim_date d ON s.date_id = d.date_id
GROUP BY
d.year, d.month, p.product_category
ORDER BY
d.year, d.month, p.product_category;

# 输出日志
year | month | product_category | total_quantity | total_amount
——+——-+——————+—————-+————–
2026 | 1 | Category 1 | 3 | 400.00
2026 | 1 | Category 2 | 1 | 300.00
2026 | 2 | Category 1 | 3 | 300.00
(3 rows)

# 查询客户销售数据
SELECT
c.customer_name,
SUM(s.quantity) AS total_quantity,
SUM(s.amount) AS total_amount
FROM
fact_sales s
JOIN
dim_customer c ON s.customer_id = c.customer_id
GROUP BY
c.customer_name
ORDER BY
total_amount DESC;

# 输出日志
customer_name | total_quantity | total_amount
—————+—————-+————–
Customer 2 | 4 | 600.00
Customer 1 | 3 | 400.00
(2 rows)

# 退出PostgreSQL
\q