# 1. 安装PostgreSQL
$ sudo yum install postgresql14 postgresql14-server

# 2. 初始化数据库
$ sudo /usr/pgsql-14/bin/postgresql-14-setup initdb

# 3. 启动PostgreSQL服务
$ sudo systemctl start postgresql
$ sudo systemctl enable postgresql

# 4. 创建ETL用户和数据库
$ sudo -u postgres psql
CREATE USER etl_user WITH PASSWORD ‘password’;
CREATE DATABASE etl_db OWNER etl_user;
\q

# 5. 配置PostgreSQL参数
$ sudo vi /postgresql/fgapp/14/data/postgresql.conf
shared_buffers = 4GB
work_mem = 32MB
maintenance_work_mem = 1GB
effective_cache_size = 12GB
max_connections = 100

# 6. 重启PostgreSQL服务
$ sudo systemctl restart postgresql

# 7. 验证环境
$ psql -U etl_user -d etl_db -c “SELECT version();”

# 1. 安装Talend Open Studio
$ wget https://downloads.talend.com/TOS-ESB-20211117_1054-V8.0.1.zip
$ unzip TOS-ESB-20211117_1054-V8.0.1.zip

# 2. 安装Python和必要的库
$ sudo yum install python3 python3-pip
$ pip3 install pandas sqlalchemy psycopg2-binary

# 3. 安装Apache Airflow
$ pip3 install apache-airflow
$ airflow db init
$ airflow users create –username admin –firstname Admin –lastname User –role Admin –email admin@fgedu.net.cn

# 4. 启动Airflow
$ airflow webserver –port 8080
$ airflow scheduler

# 5. 验证ETL工具
$ python3 -c “import pandas; print(‘pandas installed successfully’);”

# 1. 提取数据
# 使用Python脚本提取数据
$ cat extract.py
import pandas as pd
from sqlalchemy import create_engine

# 连接到源数据库
source_engine = create_engine(‘postgresql://source_user:source_password@source_server:5432/source_db’)

# 提取数据
df = pd.read_sql(‘SELECT * FROM sales’, source_engine)

# 保存提取的数据
df.to_csv(‘/tmp/sales.csv’, index=False)

$ python3 extract.py

# 2. 转换数据
# 使用Python脚本转换数据
$ cat transform.py
import pandas as pd

# 读取数据
df = pd.read_csv(‘/tmp/sales.csv’)

# 数据清洗
df = df.dropna()
df = df[df[‘amount’] > 0]

# 数据转换
df[‘sale_date’] = pd.to_datetime(df[‘sale_date’])
df[‘month’] = df[‘sale_date’].dt.month
df[‘year’] = df[‘sale_date’].dt.year

# 保存转换后的数据
df.to_csv(‘/tmp/sales_transformed.csv’, index=False)

$ python3 transform.py

# 3. 加载数据
# 使用Python脚本加载数据
$ cat load.py
import pandas as pd
from sqlalchemy import create_engine

# 连接到目标数据库
target_engine = create_engine(‘postgresql://etl_user:password@fgedu.localhost:5432/etl_db’)

# 读取转换后的数据
df = pd.read_csv(‘/tmp/sales_transformed.csv’)

# 加载数据到目标表
df.to_sql(‘sales_fact’, target_engine, if_exists=’fgappend’, index=False)

$ python3 load.py

# 4. 验证数据加载
$ psql -U etl_user -d etl_db -c “SELECT COUNT(*) FROM sales_fact;”

# 1. 批量处理
# 使用COPY命令批量加载数据
$ psql -U etl_user -d etl_db -c “COPY sales_fact FROM ‘/tmp/sales_transformed.csv’ CSV HEADER;”

# 2. 并行处理
# 使用Python多线程处理数据
$ cat parallel_transform.py
import pandas as pd
from concurrent.futures import ThreadPoolExecutor

def process_chunk(chunk):
# 处理数据块
chunk = chunk.dropna()
chunk = chunk[chunk[‘amount’] > 0]
chunk[‘sale_date’] = pd.to_datetime(chunk[‘sale_date’])
return chunk

# 读取数据
chunks = pd.read_csv(‘/tmp/sales.csv’, chunksize=10000)

# 并行处理数据
with ThreadPoolExecutor(max_workers=4) as executor:
processed_chunks = list(executor.map(process_chunk, chunks))

# 合并处理后的数据
df = pd.concat(processed_chunks)

# 保存转换后的数据
df.to_csv(‘/tmp/sales_transformed.csv’, index=False)

$ python3 parallel_transform.py

# 3. 索引优化
$ psql -U etl_user -d etl_db -c “CREATE INDEX sales_fact_sale_date_idx ON sales_fact(sale_date);”

# 4. 分区表
$ psql -U etl_user -d etl_db -c “CREATE TABLE sales_fact (id SERIAL PRIMARY KEY, sale_date DATE, amount NUMERIC) PARTITION BY RANGE (sale_date);”
$ psql -U etl_user -d etl_db -c “CREATE TABLE sales_fact_y2023 PARTITION OF sales_fact FOR VALUES FROM (‘2023-01-01’) TO (‘2024-01-01’);”

# 5. 验证优化效果
$ time python3 load.py