# 1. 安装PostgreSQL
$ yum install -y https://download.postgresql.org/pub/repos/yum/reporpms/EL-9-x86_64/pgdg-redhat-repo-latest.noarch.rpm
$ yum install -y postgresql18-server postgresql18-contrib

# 2. 初始化数据库
$ /postgresql/fgapp/bin/postgresql-18-setup initdb

# 3. 配置PostgreSQL
$ vi /postgresql/fgdata/postgresql.conf

# 内存配置
shared_buffers = 8GB # 物理内存的25%
work_mem = 64MB # 根据并发数和查询复杂度调整
maintenance_work_mem = 2GB # 用于VACUUM等操作
effective_cache_size = 24GB # 物理内存的75%

# 查询优化
random_page_cost = 1.1 # SSD存储时设置为1.1
seq_page_cost = 1.0
cpu_tuple_cost = 0.01
cpu_index_tuple_cost = 0.005
cpu_operator_cost = 0.0025

# 并行查询
max_parallel_workers_per_gather = 4
max_parallel_workers = 8

# WAL配置
wal_level = replica
max_wal_size = 1GB
min_wal_size = 80MB

# 检查点
checkpoint_completion_target = 0.9

# 4. 启动PostgreSQL
$ systemctl start postgresql-18
$ systemctl enable postgresql-18

# 1. 创建维度表
$ psql -U postgres -d fgedudb

fgedudb=# CREATE TABLE dim_date (
date_id serial PRIMARY KEY,
date_date date NOT NULL,
year integer NOT NULL,
quarter integer NOT NULL,
month integer NOT NULL,
day integer NOT NULL,
day_of_week integer NOT NULL,
is_weekend boolean NOT NULL
);
CREATE TABLE

fgedudb=# CREATE TABLE dim_customer (
customer_id serial PRIMARY KEY,
customer_name text NOT NULL,
email text NOT NULL,
phone text NOT NULL,
address text NOT NULL,
city text NOT NULL,
country text NOT NULL
);
CREATE TABLE

fgedudb=# CREATE TABLE dim_product (
product_id serial PRIMARY KEY,
product_name text NOT NULL,
category text NOT NULL,
price numeric(10,2) NOT NULL,
supplier text NOT NULL
);
CREATE TABLE

# 2. 创建事实表
fgedudb=# CREATE TABLE fact_sales (
sales_id serial PRIMARY KEY,
date_id integer NOT NULL REFERENCES dim_date(date_id),
customer_id integer NOT NULL REFERENCES dim_customer(customer_id),
product_id integer NOT NULL REFERENCES dim_product(product_id),
quantity integer NOT NULL,
amount numeric(10,2) NOT NULL,
sales_date timestamp NOT NULL
);
CREATE TABLE

# 3. 创建索引
fgedudb=# CREATE INDEX idx_fact_sales_date_id ON fact_sales (date_id);
CREATE INDEX

fgedudb=# CREATE INDEX idx_fact_sales_customer_id ON fact_sales (customer_id);
CREATE INDEX

fgedudb=# CREATE INDEX idx_fact_sales_product_id ON fact_sales (product_id);
CREATE INDEX

fgedudb=# CREATE INDEX idx_fact_sales_sales_date ON fact_sales (sales_date);
CREATE INDEX

# 1. 安装Apache Airflow
$ pip install apache-airflow

# 2. 初始化Airflow
$ airflow db init

# 3. 创建ETL DAG
$ vi /opt/airflow/dags/etl_dag.py

from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime, timedelta
import psycopg2

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

dag = DAG(
‘etl_dag’,
default_args=default_args,
description=’ETL process for data warehouse’,
schedule_interval=timedelta(days=1),
)

def extract_data():
# 连接源数据库
source_conn = psycopg2.connect(
host=’fgedu.localhost’,
port=5432,
database=’source_db’,
user=’postgres’,
password=’postgres_pass’
)
source_cur = source_conn.cursor()

# 提取数据
source_cur.execute(“SELECT * FROM sales”)
sales_data = source_cur.fetchall()

source_cur.close()
source_conn.close()

return sales_data

def transform_data(**kwargs):
ti = kwargs[‘ti’]
sales_data = ti.xcom_pull(task_ids=’extract’)

# 转换数据
transformed_data = []
for row in sales_data:
transformed_data.fgappend({
‘date_id’: row[0],
‘customer_id’: row[1],
‘product_id’: row[2],
‘quantity’: row[3],
‘amount’: row[4],
‘sales_date’: row[5]
})

return transformed_data

def load_data(**kwargs):
ti = kwargs[‘ti’]
transformed_data = ti.xcom_pull(task_ids=’transform’)

# 连接数据仓库
dw_conn = psycopg2.connect(
host=’fgedu.localhost’,
port=5432,
database=’fgedudb’,
user=’postgres’,
password=’postgres_pass’
)
dw_cur = dw_conn.cursor()

# 加载数据
for data in transformed_data:
dw_cur.execute(
“INSERT INTO fact_sales (date_id, customer_id, product_id, quantity, amount, sales_date) VALUES (%s, %s, %s, %s, %s, %s)”,
(data[‘date_id’], data[‘customer_id’], data[‘product_id’], data[‘quantity’], data[‘amount’], data[‘sales_date’])
)

dw_conn.commit()
dw_cur.close()
dw_conn.close()

extract_task = PythonOperator(
task_id=’extract’,
python_callable=extract_data,
dag=dag,
)

transform_task = PythonOperator(
task_id=’transform’,
python_callable=transform_data,
provide_context=True,
dag=dag,
)

load_task = PythonOperator(
task_id=’load’,
python_callable=load_data,
provide_context=True,
dag=dag,
)

extract_task >> transform_task >> load_task

# 4. 启动Airflow
$ airflow webserver -p 8080
$ airflow scheduler

# 1. 创建分区表
$ psql -U postgres -d fgedudb

fgedudb=# CREATE TABLE fact_sales_partitioned (
sales_id serial,
date_id integer NOT NULL REFERENCES dim_date(date_id),
customer_id integer NOT NULL REFERENCES dim_customer(customer_id),
product_id integer NOT NULL REFERENCES dim_product(product_id),
quantity integer NOT NULL,
amount numeric(10,2) NOT NULL,
sales_date timestamp NOT NULL
) PARTITION BY RANGE (sales_date);
CREATE TABLE

# 2. 创建分区
fgedudb=# CREATE TABLE fact_sales_2025_01 PARTITION OF fact_sales_partitioned
FOR VALUES FROM (‘2025-01-01’) TO (‘2025-02-01’);
CREATE TABLE

fgedudb=# CREATE TABLE fact_sales_2025_02 PARTITION OF fact_sales_partitioned
FOR VALUES FROM (‘2025-02-01’) TO (‘2025-03-01’);
CREATE TABLE

# 3. 创建索引
fgedudb=# CREATE INDEX idx_fact_sales_partitioned_date_id ON fact_sales_partitioned (date_id);
CREATE INDEX

fgedudb=# CREATE INDEX idx_fact_sales_partitioned_customer_id ON fact_sales_partitioned (customer_id);
CREATE INDEX

fgedudb=# CREATE INDEX idx_fact_sales_partitioned_product_id ON fact_sales_partitioned (product_id);
CREATE INDEX

# 4. 测试查询性能
fgedudb=# EXPLAIN ANALYZE SELECT * FROM fact_sales_partitioned WHERE sales_date BETWEEN ‘2025-01-01’ AND ‘2025-01-31’;

# 1. 创建物化视图
$ psql -U postgres -d fgedudb

fgedudb=# CREATE MATERIALIZED VIEW mv_sales_by_month AS
SELECT
d.year,
d.month,
SUM(f.amount) AS total_sales,
COUNT(*) AS order_count
FROM fact_sales f
JOIN dim_date d ON f.date_id = d.date_id
GROUP BY d.year, d.month
ORDER BY d.year, d.month;
CREATE MATERIALIZED VIEW

# 2. 刷新物化视图
fgedudb=# REFRESH MATERIALIZED VIEW mv_sales_by_month;
REFRESH MATERIALIZED VIEW

# 3. 创建索引
fgedudb=# CREATE INDEX idx_mv_sales_by_month_year_month ON mv_sales_by_month (year, month);
CREATE INDEX

# 4. 测试查询性能
fgedudb=# EXPLAIN ANALYZE SELECT * FROM mv_sales_by_month WHERE year = 2025 AND month = 1;