# 1. 数据抽取（Extract）
# 使用Python从MySQL抽取数据
import pymysql
from pymongo import MongoClient
import json

# 连接MySQL
mysql_conn = pymysql.connect(
host=’192.168.1.100′,
port=3306,
user=’fgedu’,
password=’fgedu123′,
database=’fgedudb’
)

# 连接MongoDB
mongo_client = MongoClient(‘mongodb://fgedu:fgedu123@192.168.1.101:27017/admin’)
mongo_db = mongo_client[‘fgedu_warehouse’]

# 抽取数据
with mysql_conn.cursor() as cursor:
cursor.execute(“SELECT * FROM fgedu_users”)
results = cursor.fetchall()

# 转换数据
for row in results:
doc = {
‘user_id’: row[0],
‘name’: row[1],
‘age’: row[2],
’email’: row[3],
‘created_at’: row[4]
}
# 加载到MongoDB
mongo_db.fgedu_users_ods.insert_one(doc)

# 2. 数据转换（Transform）
# 使用Aggregation Pipeline清洗数据
pipeline = [
{
‘$project’: {
‘user_id’: 1,
‘name’: 1,
‘age’: 1,
’email’: 1,
‘created_at’: 1,
‘age_group’: {
‘$switch’: {
‘branches’: [
{ ‘case’: { ‘$lt’: [‘$age’, 18] }, ‘then’: ‘未成年’ },
{ ‘case’: { ‘$lt’: [‘$age’, 35] }, ‘then’: ‘青年’ },
{ ‘case’: { ‘$lt’: [‘$age’, 50] }, ‘then’: ‘中年’ },
{ ‘case’: { ‘$gte’: [‘$age’, 50] }, ‘then’: ‘老年’ }
],
‘default’: ‘未知’
}
}
}
},
{
‘$merge’: {
‘into’: ‘fgedu_users_dwd’,
‘on’: ‘user_id’,
‘whenMatched’: ‘merge’,
‘whenNotMatched’: ‘insert’
}
}
]
mongo_db.fgedu_users_ods.aggregate(pipeline)

# 3. 数据加载（Load）
# 使用$out将聚合结果输出到新集合
pipeline = [
{
‘$group’: {
‘_id’: ‘$age_group’,
‘count’: { ‘$sum’: 1 },
‘avg_age’: { ‘$avg’: ‘$age’ }
}
},
{
‘$out’: ‘fgedu_users_dws’
}
]
mongo_db.fgedu_users_dwd.aggregate(pipeline)