数据管理解决方案

# 加载必要的包
library(dplyr)
library(lubridate)
library(stringr)
library(DBI)
library(RMySQL)

# 数据库连接配置
con <- dbConnect(MySQL(),
                 host = "localhost",
                 dbname = "hospital_db",
                 username = "data_analyst",
                 password = "secure_password")

# 数据提取函数
extract_patient_data <- function(start_date, end_date) {
  query <- paste0("
    SELECT 
      p.patient_id,
      p.name,
      p.birth_date,
      p.gender,
      v.visit_date,
      v.department,
      v.diagnosis,
      l.test_name,
      l.test_value,
      l.test_unit
    FROM patients p
    LEFT JOIN visits v ON p.patient_id = v.patient_id
    LEFT JOIN lab_results l ON v.visit_id = l.visit_id
    WHERE v.visit_date BETWEEN '", start_date, "' AND '", end_date, "'
  ")
  
  dbGetQuery(con, query)
}

# 数据清洗函数
clean_patient_data <- function(raw_data) {
  cleaned_data <- raw_data %>%
    # 处理缺失值
    filter(!is.na(patient_id)) %>%
    # 标准化性别字段
    mutate(
      gender = case_when(
        str_detect(gender, "^[男M]") ~ "M",
        str_detect(gender, "^[女F]") ~ "F",
        TRUE ~ "Unknown"
      ),
      # 计算年龄
      age = as.numeric(difftime(visit_date, birth_date, units = "days")) / 365.25,
      # 标准化科室名称
      department = str_trim(str_to_upper(department)),
      # 处理检验值异常
      test_value = ifelse(test_value < 0 | test_value > 10000, NA, test_value)
    ) %>%
    # 去除重复记录
    distinct()
  
  return(cleaned_data)
}

# 数据质量检查
quality_check <- function(data) {
  quality_report <- list(
    total_records = nrow(data),
    missing_patient_id = sum(is.na(data$patient_id)),
    missing_visit_date = sum(is.na(data$visit_date)),
    invalid_age = sum(data$age < 0 | data$age > 120, na.rm = TRUE),
    duplicate_records = nrow(data) - nrow(distinct(data))
  )
  
  # 计算数据质量得分
  quality_score <- (1 - (quality_report$missing_patient_id + 
                        quality_report$missing_visit_date + 
                        quality_report$invalid_age + 
                        quality_report$duplicate_records) / 
                        quality_report$total_records) * 100
  
  quality_report$quality_score <- round(quality_score, 2)
  
  return(quality_report)
}

# 执行数据处理流程
process_daily_data <- function(date) {
  # 提取数据
  raw_data <- extract_patient_data(date, date)
  
  # 清洗数据
  clean_data <- clean_patient_data(raw_data)
  
  # 质量检查
  quality_report <- quality_check(clean_data)
  
  # 输出质量报告
  cat("数据质量报告 - ", date, "\n")
  cat("总记录数:", quality_report$total_records, "\n")
  cat("数据质量得分:", quality_report$quality_score, "%\n")
  
  return(list(data = clean_data, quality = quality_report))
}

# 示例：处理今日数据
today_result <- process_daily_data(Sys.Date())
                    

# 数据分层管理
# ODS层 (Operational Data Store) - 原始数据存储
# DWD层 (Data Warehouse Detail) - 明细数据层
# DWS层 (Data Warehouse Summary) - 汇总数据层
# ADS层 (Application Data Service) - 应用数据服务层

# 创建分层表结构
create_ods_tables <- function() {
  # ODS层 - 患者原始数据
  dbExecute(con, "
    CREATE TABLE IF NOT EXISTS ods_patients (
      patient_id VARCHAR(20) PRIMARY KEY,
      name VARCHAR(100),
      birth_date DATE,
      gender CHAR(1),
      phone VARCHAR(20),
      address TEXT,
      created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
      updated_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
    )
  ")
  
  # DWD层 - 清洗后的明细数据
  dbExecute(con, "
    CREATE TABLE IF NOT EXISTS dwd_patient_visits (
      visit_id VARCHAR(30) PRIMARY KEY,
      patient_id VARCHAR(20),
      visit_date DATE,
      department VARCHAR(50),
      doctor_id VARCHAR(20),
      diagnosis TEXT,
      age_at_visit INT,
      visit_type VARCHAR(20),
      created_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
  ")
  
  # DWS层 - 按科室汇总的统计数据
  dbExecute(con, "
    CREATE TABLE IF NOT EXISTS dws_department_daily (
      stat_date DATE,
      department VARCHAR(50),
      total_visits INT,
      unique_patients INT,
      avg_age DECIMAL(5,2),
      male_ratio DECIMAL(5,4),
      PRIMARY KEY (stat_date, department)
    )
  ")
}

# 数据质量监控
monitor_data_quality <- function() {
  # 检查数据完整性
  completeness_check <- dbGetQuery(con, "
    SELECT 
      'patients' as table_name,
      COUNT(*) as total_records,
      SUM(CASE WHEN patient_id IS NULL THEN 1 ELSE 0 END) as null_patient_id,
      SUM(CASE WHEN name IS NULL OR name = '' THEN 1 ELSE 0 END) as null_name,
      SUM(CASE WHEN birth_date IS NULL THEN 1 ELSE 0 END) as null_birth_date
    FROM ods_patients
  ")
  
  # 检查数据一致性
  consistency_check <- dbGetQuery(con, "
    SELECT 
      COUNT(*) as total_visits,
      COUNT(DISTINCT patient_id) as unique_patients,
      SUM(CASE WHEN age_at_visit < 0 OR age_at_visit > 120 THEN 1 ELSE 0 END) as invalid_age
    FROM dwd_patient_visits
    WHERE visit_date = CURDATE()
  ")
  
  return(list(
    completeness = completeness_check,
    consistency = consistency_check
  ))
}