ETL 数据工程

-- 从 MES 系统的 production_log 表抽取昨天的 OEE 数据
SELECT
    machine_id,
    shift_date,
    availability_rate   -- 可用率
    performance_rate    -- 性能率
    quality_rate        -- 良品率
    FROM mes_database.dbo.production_log
WHERE shift_date = CAST(GETDATE() - 1 AS DATE);

# 读取设备停机日志 CSV 文件
import pandas as pd

df_downtime = pd.read_csv(
    'equipment_downtime_2024.csv',
    encoding='utf-8',          # 中文文件注意编码
    parse_dates=['downtime_start', 'downtime_end']
)
print(df_downtime.head())

# 读取质检系统导出的 Excel，指定 sheet
df_qc = pd.read_excel(
    'quality_report_jan.xlsx',
    sheet_name='测试数据',
    header=1                  # 第2行是表头
)

import requests

response = requests.get(
    'http://erp.company.com/api/workorders',
    params={'date': '2024-01-15'},
    headers={'Authorization': 'Bearer xxx'}
)
df_orders = pd.DataFrame(response.json()['data'])

-- 方法1：用全局平均值填充 NULL
SELECT
    machine_id,
    COALESCE(oee, (SELECT AVG(oee) FROM production_log WHERE oee IS NOT NULL)) AS oee
FROM production_log;

-- 方法2：用前一行填充 (SQL Server)
SELECT
    machine_id,
    LAST_VALUE(temp IGNORE NULLS)
        OVER (ORDER BY machine_id
              ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS temp
FROM production_log;

# 查看缺失情况
print(df.isnull().sum())

# 方法1：用平均值填充
df['oee'].fillna(df['oee'].mean(), inplace=True)

# 方法2：用前一个有效值填充（适合时序传感器数据）
df['temp'].fillna(method='ffill', inplace=True)

# 方法3：直接删掉整行（缺失太多时）
df.dropna(subset=['machine_id'], inplace=True)

# 温度传感器：正常范围 20~80°C
valid_mask = (df['temp'] >= 20) & (df['temp'] <= 80)
mean_temp = df.loc[valid_mask, 'temp'].mean()

# 将异常值替换为正常数据的平均值
df.loc[~valid_mask, 'temp'] = mean_temp

# 或者直接过滤掉异常行
df = df[valid_mask].copy()

# "85.6%" → 85.6
df['oee'] = df['oee_str'].str.replace('%', '').astype(float)

# "1,234" → 1234
df['quantity'] = df['quantity'].str.replace(',', '').astype(int)

# pandas 自动解析多种日期格式
df['date'] = pd.to_datetime(df['date'], errors='coerce')

# 统一输出为 "2024-01-15" 格式
df['date'] = df['date'].dt.strftime('%Y-%m-%d')

# errors='coerce'：解析失败的变为 NaT（而不是报错）

-- SQL Server: 各种格式统一转 DATE
SELECT
    CAST(CASE
        WHEN ISDATE(raw_date) = 1 THEN raw_date
        ELSE NULL
    END AS DATE) AS clean_date
FROM staging_table;

-- 格式化输出
SELECT FORMAT(clean_date, 'yyyy-MM-dd') AS date_str;

-- 使用 ROW_NUMBER() 保留最新的一条
WITH ranked AS (
    SELECT *,
        ROW_NUMBER() OVER (
            PARTITION BY machine_id, shift_date
            ORDER BY update_time DESC
        ) AS rn
    FROM production_log
)
SELECT * FROM ranked WHERE rn = 1;

# 按 machine_id + date 去重，保留最后一条（最新的）
df = df.drop_duplicates(
    subset=['machine_id', 'date'],
    keep='last'
)

# 系统A 的中文列名 → 统一英文列名
df_mes = df_mes.rename(columns={
    '设备号': 'machine_id',
    '班次日期': 'shift_date',
    '可用率': 'availability_rate',
})

# 现在两个 DataFrame 列名一致，可以 merge 了
df_merged = pd.merge(df_mes, df_erp, on=['machine_id', 'shift_date'])

-- 全量加载：先清空，再插入
TRUNCATE TABLE oee_summary;
INSERT INTO oee_summary (machine_id, shift_date, oee)
SELECT machine_id, shift_date, oee
FROM stg_production_clean;

-- 增量加载：只插入不存在的新数据
INSERT INTO oee_summary (machine_id, shift_date, oee)
SELECT s.machine_id, s.shift_date, s.oee
FROM stg_production_clean s
LEFT JOIN oee_summary t
    ON s.machine_id = t.machine_id AND s.shift_date = t.shift_date
WHERE t.machine_id IS NULL;  -- 目标表里没有的

-- 增量更新：有的更新，没有的新增
MERGE INTO oee_summary AS target
USING stg_production_clean AS source
    ON target.machine_id = source.machine_id
    AND target.shift_date = source.shift_date

-- 匹配到了 → 更新
WHEN MATCHED THEN
    UPDATE SET
        target.oee = source.oee,
        target.update_time = GETDATE()

-- 没匹配到 → 插入
WHEN NOT MATCHED THEN
    INSERT (machine_id, shift_date, oee, update_time)
    VALUES (source.machine_id, source.shift_date, source.oee, GETDATE());

-- 每台设备的 OEE 在所有设备中的排名
SELECT
    machine_id,
    shift_date,
    oee,
    RANK() OVER (ORDER BY oee DESC) AS oee_rank,
    -- 7天移动平均 OEE（看趋势）
    AVG(oee) OVER (
        PARTITION BY machine_id
        ORDER BY shift_date
        ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
    ) AS oee_7day_avg,
    -- 当天在同设备中的 OEE 占比
    oee * 1.0 / SUM(oee) OVER (PARTITION BY shift_date) AS oee_ratio
FROM production_log
WHERE shift_date >= '2024-01-01';

MERGE INTO <目标表> AS T
USING <源表/子查询> AS S
    ON T.主键 = S.主键
WHEN MATCHED THEN UPDATE SET ...
WHEN NOT MATCHED THEN INSERT ...
WHEN NOT MATCHED BY SOURCE THEN DELETE;  -- 可选：删掉源里没有的

CREATE PROCEDURE sp_daily_etl
    @target_date DATE = NULL
AS
BEGIN
    SET NOCOUNT ON;

    -- 默认处理昨天
    IF @target_date IS NULL
        SET @target_date = CAST(GETDATE() - 1 AS DATE);

    -- Step 1: 抽取到临时表
    SELECT machine_id, shift_date, oee, temp
    INTO #staging
    FROM mes_database.dbo.production_log
    WHERE shift_date = @target_date;

    -- Step 2: 清洗（去NULL、去异常）
    DELETE FROM #staging WHERE oee IS NULL OR oee > 100;

    -- Step 3: 增量写入目标表
    MERGE INTO oee_summary AS T
    USING #staging AS S
        ON T.machine_id = S.machine_id AND T.shift_date = S.shift_date
    WHEN MATCHED THEN UPDATE SET
        T.oee = S.oee, T.temp = S.temp, T.update_time = GETDATE()
    WHEN NOT MATCHED THEN INSERT
        (machine_id, shift_date, oee, temp, update_time)
        VALUES (S.machine_id, S.shift_date, S.oee, S.temp, GETDATE());

    -- 记录执行日志
    INSERT INTO etl_log (job_name, run_date, rows_processed, status)
    VALUES ('daily_oee_etl', @target_date, @@ROWCOUNT, 'SUCCESS');

    DROP TABLE #staging;
END

-- 创建作业
EXEC msdb.dbo.sp_add_job
    @job_name = N'Daily_OEE_ETL';

-- 添加步骤
EXEC msdb.dbo.sp_add_jobstep
    @job_name = N'Daily_OEE_ETL',
    @step_name = N'Run ETL',
    @subsystem = N'TSQL',
    @command = N'EXEC sp_daily_etl;';

-- 设置调度：每天早上6:00
EXEC msdb.dbo.sp_add_jobschedule
    @job_name = N'Daily_OEE_ETL',
    @name = N'Daily 6AM',
    @freq_type = 4,          -- 每天
    @active_start_time = 60000; -- 06:00:00

import pandas as pd
from sqlalchemy import create_engine

# =============================================
# EXTRACT — 从 3 个数据源抽取
# =============================================

# 1. MES 系统导出的 OEE Excel
df_oee = pd.read_excel('mes_oee_jan.xlsx')

# 2. 设备停机日志 CSV
df_downtime = pd.read_csv(
    'downtime_log_jan.csv',
    encoding='gbk',
    parse_dates=['start_time', 'end_time']
)

# 3. 质检报表 Excel（指定 sheet）
df_qc = pd.read_excel('quality_jan.xlsx', sheet_name='成品测试')

# =============================================
# TRANSFORM — 清洗 + 转换
# =============================================

# --- 清洗 OEE 数据 ---
# 列名映射：中文 → 英文
df_oee = df_oee.rename(columns={
    '设备编号': 'machine_id',
    '日期': 'shift_date',
    'OEE': 'oee'
})

# 去掉 OEE 中的 "%"，转 float
df_oee['oee'] = df_oee['oee'].astype(str).str.replace('%', '')
df_oee['oee'] = pd.to_numeric(df_oee['oee'], errors='coerce')

# 过滤异常值（OEE > 100 或 < 0 的不正常）
df_oee = df_oee[(df_oee['oee'] >= 0) & (df_oee['oee'] <= 100)]

# 填充空值
df_oee['oee'].fillna(df_oee['oee'].mean(), inplace=True)

# 统一日期格式
df_oee['shift_date'] = pd.to_datetime(df_oee['shift_date']).dt.date

# --- 清洗停机数据 ---
df_downtime = df_downtime.rename(columns={
    'machine_id': 'machine_id',
    'down_minutes': 'downtime_min'
})
# 按设备+日期汇总停机时长
df_downtime['shift_date'] = df_downtime['start_time'].dt.date
df_down_sum = df_downtime.groupby(['machine_id', 'shift_date'])
    ['downtime_min'].sum().reset_index()

# --- 清洗质检数据 ---
df_qc = df_qc.rename(columns={
    '设备': 'machine_id',
     '检测日期': 'shift_date',
    '不良率': 'defect_rate'
})
df_qc['shift_date'] = pd.to_datetime(df_qc['shift_date']).dt.date

# =============================================
# MERGE — 关联合并
# =============================================

# 三张表按 machine_id + shift_date 关联
df_merged = df_oee.merge(df_down_sum, on=['machine_id', 'shift_date'], how='left')
df_merged = df_merged.merge(df_qc[['machine_id', 'shift_date', 'defect_rate']],
                      on=['machine_id', 'shift_date'], how='left')

# 去重
df_merged = df_merged.drop_duplicates(subset=['machine_id', 'shift_date'])

# =============================================
# LOAD — 写入 SQL Server
# =============================================

engine = create_engine(
    'mssql+pyodbc://user:pass@server/db?driver=ODBC+Driver+17+for+SQL+Server'
)

# 写入目标表（有则替换）
df_merged.to_sql(
    'production_daily',
    engine,
    if_exists='append',    # 追加模式
    index=False
)

print(f'ETL 完成！共写入 {len(df_merged)} 行')

# 编辑定时任务
# crontab -e

# 格式: 分 时 日 月 星期  命令
# ┌─────── 分钟 (0-59)
# │ ┌───── 小时 (0-23)
# │ │ ┌─── 日 (1-31)
# │ │ │ ┌─ 月 (1-12)
# │ │ │ │ ┌ 星期 (0-6)

# 每天早上 6:00 执行 ETL
0 6 * * * /usr/bin/python3 /opt/etl/daily_etl.py >> /var/log/etl.log 2>&1

# 每30分钟执行一次
*/30 * * * * /usr/bin/python3 /opt/etl/realtime_sync.py

# 工作日早上 7:00
0 7 * * 1-5 /usr/bin/python3 /opt/etl/weekday_report.py

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

with DAG(
    dag_id='manufacturing_daily_etl',
    schedule='0 6 * * *',       # 每天 6:00
    start_date=datetime(2024, 1, 1),
    catchup=False
) as dag:

    # 定义 3 个任务，按顺序执行
    extract = PythonOperator(task_id='extract', python_callable=run_extract)
    transform = PythonOperator(task_id='transform', python_callable=run_transform)
    load = PythonOperator(task_id='load', python_callable=run_load)

    # 任务依赖：extract → transform → load
    extract >> transform >> load

CREATE PROCEDURE sp_data_quality_check
AS
BEGIN
    -- 1. 行数对比
    DECLARE @src_count INT, @tgt_count INT;
    SELECT @src_count = COUNT(*) FROM production_log WHERE shift_date = CAST(GETDATE()-1 AS DATE);
    SELECT @tgt_count = COUNT(*) FROM oee_summary WHERE shift_date = CAST(GETDATE()-1 AS DATE);

    IF ABS(@src_count - @tgt_count) > @src_count * 0.05  -- 差异超过5%
        INSERT INTO dq_alerts VALUES ('ROW_COUNT_MISMATCH', GETDATE());

    -- 2. 范围检查
    IF EXISTS (SELECT 1 FROM oee_summary WHERE oee > 100 OR oee < 0)
        INSERT INTO dq_alerts VALUES ('OEE_OUT_OF_RANGE', GETDATE());

    -- 3. 空值检查
    IF EXISTS (SELECT 1 FROM oee_summary WHERE machine_id IS NULL)
        INSERT INTO dq_alerts VALUES ('NULL_MACHINE_ID', GETDATE());

    -- 4. 唯一性检查
    IF EXISTS (
        SELECT machine_id, shift_date FROM oee_summary
        GROUP BY machine_id, shift_date HAVING COUNT(*) > 1
    )
        INSERT INTO dq_alerts VALUES ('DUPLICATE_KEY', GETDATE());
END

def quality_check(df, table_name):
    issues = []

    # 空值检查
    null_counts = df.isnull().sum()
    for col, cnt in null_counts.items():
        if cnt > 0:
            issues.append(f"[{table_name}] {col} 有 {cnt} 个空值")

    # 范围检查
    if 'oee' in df.columns:
        bad_oee = df[(df['oee'] < 0) | (df['oee'] > 100)]
        if len(bad_oee) > 0:
            issues.append(f"[{table_name}] {len(bad_oee)} 条 OEE 超出范围")

    # 唯一性检查
    if 'machine_id' in df.columns:
        dupes = df.duplicated(subset=['machine_id', 'shift_date']).sum()
        if dupes > 0:
            issues.append(f"[{table_name}] {dupes} 条重复记录")

    return issues

# 使用
problems = quality_check(df_merged, 'production_daily')
if problems:
    for p in problems:
        print(f"⚠️ {p}")
else:
    print("✅ 数据质量检查通过")