+--------------------------------------------------+
|                 用户画像核心价值                    |
+--------------------------------------------------+
|                                                    |
|  +------------+  +------------+  +------------+   |
|  | 精准营销   |  | 个性化推荐 |  | 风险控制   |   |
|  +------------+  +------------+  +------------+   |
|  | 定向投放   |  | 商品推荐   |  | 欺诈检测   |   |
|  | 差异化定价 |  | 内容推荐   |  | 信用评估   |   |
|  | 活动触达   |  | 搜索排序   |  | 流失预警   |   |
|  +------------+  +------------+  +------------+   |
|                                                    |
|  +------------+  +------------+                    |
|  | 产品优化   |  | 运营决策   |                    |
|  +------------+  +------------+                    |
|  | 功能改进   |  | 资源配置   |                    |
|  | 体验提升   |  | 战略规划   |                    |
|  +------------+  +------------+                    |
+--------------------------------------------------+

+-----------------------------------------------------------------+
|                        标签体系分类                               |
+-----------------------------------------------------------------+
|                                                                   |
|  +------------------+  +------------------+  +----------------+  |
|  |   事实标签       |  |   统计标签       |  |   算法标签     |  |
|  +------------------+  +------------------+  +----------------+  |
|  | 直接从原始数据   |  | 对事实标签进行   |  | 通过机器学习   |  |
|  | 中提取的客观     |  | 统计计算得到的   |  | 算法挖掘得到   |  |
|  | 事实描述         |  | 聚合指标         |  | 的预测标签     |  |
|  +------------------+  +------------------+  +----------------+  |
|  | - 性别: 男/女    |  | - 消费总金额     |  | - 购买倾向     |  |
|  | - 年龄: 25岁     |  | - 购买频次       |  | - 流失概率     |  |
|  | - 注册日期       |  | - 平均客单价     |  | - 信用等级     |  |
|  | - 所在城市       |  | - 最近购买天数   |  | - 用户分群     |  |
|  +------------------+  +------------------+  +----------------+  |
+-----------------------------------------------------------------+

一级标签（大类）
├── 二级标签（中类）
│   ├── 三级标签（小类/具体标签）
│   ├── 三级标签
│   └── 三级标签
├── 二级标签
│   ├── 三级标签
│   └── 三级标签
└── 二级标签

+-------------------------------------------------------+
|                    RFM模型                              |
+-------------------------------------------------------+
|                                                         |
|  R - Recency（最近消费时间）                             |
|  ├── 用户最近一次消费距今天数                            |
|  ├── R值越小，说明用户最近刚消费过，活跃度越高           |
|  └── 反映用户的当前活跃状态                             |
|                                                         |
|  F - Frequency（消费频率）                              |
|  ├── 用户在统计周期内的消费次数                         |
|  ├── F值越大，说明用户消费越频繁，忠诚度越高            |
|  └── 反映用户的消费习惯和忠诚度                         |
|                                                         |
|  M - Monetary（消费金额）                               |
|  ├── 用户在统计周期内的消费总金额                       |
|  ├── M值越大，说明用户消费能力越强，贡献度越高          |
|  └── 反映用户的消费能力和贡献价值                       |
+-------------------------------------------------------+

+--------------------------------------------------------------------+
|                     用户画像构建流程                                 |
+--------------------------------------------------------------------+
|                                                                      |
|  [1.数据采集] --> [2.数据清洗] --> [3.标签计算] --> [4.画像存储] --> [5.画像服务]
|                                                                      |
|  +-----------+   +-----------+   +-----------+   +-----------+      +-----------+
|  | 数据采集  |   | 数据清洗  |   | 标签计算  |   | 画像存储  |      | 画像服务  |
|  +-----------+   +-----------+   +-----------+   +-----------+      +-----------+
|  | - 日志采集|   | - 去重    |   | - 事实标签|   | - Redis   |      | - REST API|
|  | - 业务数据|   | - 缺失值  |   | - 统计标签|   | - HBase   |      | - 查询服务|
|  | - 第三方  |   | - 异常值  |   | - 算法标签|   | - MySQL   |      | - 推送服务|
|  | - 爬虫    |   | - 格式化  |   | - RFM模型 |   | - Hive    |      | - 降级方案|
|  +-----------+   +-----------+   +-----------+   +-----------+      +-----------+
|                                                                      |
+--------------------------------------------------------------------+

+--------------------------------------------------------------------+
|                    画像数据存储架构                                   |
+--------------------------------------------------------------------+
|                                                                      |
|  +------------------+                                                |
|  |    Redis 7.0+    |  热数据层：实时查询、高并发访问               |
|  |    (热数据)      |  - 用户实时标签缓存                           |
|  |                  |  - 用户RFM评分                                |
|  |                  |  - 最近行为特征                               |
|  |    TTL: 24h      |  - 读写延迟 < 1ms                            |
|  +------------------+                                                |
|           |                                                          |
|           v                                                          |
|  +------------------+                                                |
|  |   HBase 2.5.x    |  标签数据层：海量标签存储、快速检索           |
|  |   (标签数据)     |  - 全量用户标签数据                           |
|  |                  |  - 标签历史版本                               |
|  |                  |  - 支持按RowKey快速查询                       |
|  +------------------+                                                |
|           |                                                          |
|           v                                                          |
|  +------------------+                                                |
|  |   MySQL 8.0      |  结构化数据层：关系查询、事务支持             |
|  |   (结构化数据)   |  - 用户基础信息                               |
|  |                  |  - 标签元数据定义                             |
|  |                  |  - 画像统计报表                               |
|  +------------------+                                                |
|           |                                                          |
|           v                                                          |
|  +------------------+                                                |
|  |   Hive 3.1.x     |  历史数据层：离线分析、数据回溯               |
|  |   (历史数据)     |  - 用户画像历史快照                           |
|  |                  |  - 标签计算中间结果                           |
|  |                  |  - 离线分析报表                               |
|  +------------------+                                                |
|                                                                      |
+--------------------------------------------------------------------+

+----------------+     +------------+     +------------+     +------------+
| 数据采集层     | --> | 数据存储层 | --> | 数据处理层 | --> | 应用层     |
+----------------+     +------------+     +------------+     +------------+
| - Kafka 3.7.0  |     | - HDFS     |     | - Spark    |     | - 推荐系统 |
| - Flume        |     | - HBase    |     |   3.5.8    |     | - 精准营销 |
| - Logstash     |     | - MySQL    |     | - Hive     |     | - 个性化   |
|                |     |   8.0      |     | - MLlib    |     | - 风控系统 |
|                |     | - Redis    |     +------------+     +------------+
|                |     |   7.0+     |
+----------------+     +------------+

ecommerce-user-profile/
├── pom.xml                              # Maven项目配置
├── src/
│   ├── main/
│   │   ├── resources/
│   │   │   ├── application.conf         # 配置文件（外部化配置）
│   │   │   └── log4j2.xml              # 日志配置
│   │   └── scala/
│   │       └── com/
│   │           └── ecommerce/
│   │               └── profile/
│   │                   ├── config/
│   │                   │   └── ConfigManager.scala      # 配置管理
│   │                   ├── dal/
│   │                   │   └── DataAccessLayer.scala     # 数据访问层
│   │                   ├── service/
│   │                   │   └── UserProfileService.scala  # 画像服务
│   │                   ├── producer/
│   │                   │   └── UserBehaviorProducer.scala # 数据采集
│   │                   ├── storage/
│   │                   │   └── UserBehaviorToHBase.scala  # HBase存储
│   │                   ├── builder/
│   │                   │   ├── UserProfileBuilder.scala   # 基础画像构建
│   │                   │   ├── RFMAnalyzer.scala          # RFM分析
│   │                   │   └── UserSegmentation.scala     # 用户分群
│   │                   ├── recommendation/
│   │                   │   └── PersonalizedRecommendation.scala # 个性化推荐
│   │                   └── marketing/
│   │                       └── TargetedMarketing.scala    # 精准营销
├── config/
│   └── application.conf                 # 外部配置文件（生产环境覆盖）
└── scripts/
    └── init_tables.sql                  # 数据库初始化脚本

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
         http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.ecommerce</groupId>
    <artifactId>user-profile</artifactId>
    <version>1.0.0</version>
    <packaging>jar</packaging>

    <name>E-Commerce User Profile System</name>
    <description>基于Spark的电商用户画像系统</description>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <scala.version>2.13.8</scala.version>
        <scala.binary.version>2.13</scala.binary.version>
        <spark.version>3.5.8</spark.version>
        <hadoop.version>3.3.6</hadoop.version>
        <hbase.version>2.5.5</hbase.version>
        <kafka.version>3.7.0</kafka.version>
        <jedis.version>5.0.2</jedis.version>
        <mysql.connector.version>8.0.33</mysql.connector.version>
        <typesafe.config.version>1.4.3</typesafe.config.version>
        <slf4j.version>2.0.9</slf4j.version>
        <log4j.version>2.21.1</log4j.version>
        <maven.compiler.source>11</maven.compiler.source>
        <maven.compiler.target>11</maven.compiler.target>
    </properties>

    <dependencies>
        <!-- Scala -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>

        <!-- Spark Core -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- Spark SQL -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- Spark Streaming -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- Spark Streaming Kafka -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-10_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <!-- Spark SQL Kafka -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql-kafka-0-10_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <!-- Spark MLlib -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-mllib_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- Spark Hive -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- Hadoop Common -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-common</artifactId>
            <version>${hadoop.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- Hadoop Client -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
            <scope>provided</scope>
        </dependency>

        <!-- HBase Client -->
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>${hbase.version}</version>
        </dependency>

        <!-- HBase Common -->
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-common</artifactId>
            <version>${hbase.version}</version>
        </dependency>

        <!-- Kafka Clients -->
        <dependency>
            <groupId>org.apache.kafka</groupId>
            <artifactId>kafka-clients</artifactId>
            <version>${kafka.version}</version>
        </dependency>

        <!-- Redis Jedis -->
        <dependency>
            <groupId>redis.clients</groupId>
            <artifactId>jedis</artifactId>
            <version>${jedis.version}</version>
        </dependency>

        <!-- MySQL Connector -->
        <dependency>
            <groupId>com.mysql</groupId>
            <artifactId>mysql-connector-j</artifactId>
            <version>${mysql.connector.version}</version>
        </dependency>

        <!-- Typesafe Config -->
        <dependency>
            <groupId>com.typesafe</groupId>
            <artifactId>config</artifactId>
            <version>${typesafe.config.version}</version>
        </dependency>

        <!-- SLF4J API -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${slf4j.version}</version>
        </dependency>

        <!-- Log4j2 -->
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-core</artifactId>
            <version>${log4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j2-impl</artifactId>
            <version>${log4j.version}</version>
        </dependency>

        <!-- ScalaTest -->
        <dependency>
            <groupId>org.scalatest</groupId>
            <artifactId>scalatest_${scala.binary.version}</artifactId>
            <version>3.2.17</version>
            <scope>test</scope>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!-- Scala Maven Plugin -->
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>4.8.1</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
                <configuration>
                    <scalaVersion>${scala.version}</scalaVersion>
                    <args>
                        <arg>-target:jvm-11</arg>
                        <arg>-deprecation</arg>
                        <arg>-feature</arg>
                        <arg>-unchecked</arg>
                    </args>
                </configuration>
            </plugin>

            <!-- Maven Shade Plugin（打包Fat Jar） -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.5.1</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>

            <!-- Maven Compiler Plugin -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.11.0</version>
                <configuration>
                    <source>11</source>
                    <target>11</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

package com.ecommerce.profile.config

import com.typesafe.config.{Config, ConfigFactory}
import org.slf4j.{Logger, LoggerFactory}

import scala.jdk.CollectionConverters._
import scala.util.Try

/**
 * 配置管理器
 *
 * 使用 Typesafe Config 库管理应用配置，支持：
 * - 多环境配置（开发/测试/生产）
 * - 环境变量覆盖（敏感信息不硬编码）
 * - 配置热加载
 * - 配置验证
 *
 * 设计原则：
 * - 所有敏感信息（密码等）通过环境变量注入，不在代码或配置文件中硬编码
 * - 提供类型安全的配置读取方法
 * - 提供默认值机制，避免因缺少配置导致启动失败
 */
object ConfigManager {

  private val logger: Logger = LoggerFactory.getLogger(ConfigManager.getClass)

  // 加载配置，优先使用环境变量覆盖
  private val config: Config = {
    val baseConfig = ConfigFactory.load()
    // 允许通过环境变量覆盖配置项
    // 例如：MYSQL_PASSWORD 环境变量会覆盖 mysql.password 配置
    val envOverrides = ConfigFactory.systemEnvironment()
    envOverrides.withFallback(baseConfig).resolve()
  }

  /**
   * 获取原始 Config 对象
   */
  def getConfig: Config = config

  // ==================== MySQL 配置 ====================

  def getMySQLUrl: String = config.getString("mysql.url")

  def getMySQLUser: String = config.getString("mysql.user")

  def getMySQLPassword: String = {
    // 优先从环境变量获取密码
    sys.env.getOrElse("MYSQL_PASSWORD", {
      if (config.hasPath("mysql.password")) {
        val pwd = config.getString("mysql.password")
        if (pwd == null || pwd.isEmpty) {
          logger.warn("MySQL密码未配置，请设置环境变量 MYSQL_PASSWORD 或配置 mysql.password")
        }
        pwd
      } else {
        logger.error("MySQL密码未配置！请设置环境变量 MYSQL_PASSWORD")
        ""
      }
    })
  }

  def getMySQLDriver: String = config.getString("mysql.driver")

  def getMySQLPoolSize: Int = Try(config.getInt("mysql.pool-size")).getOrElse(10)

  // ==================== Redis 配置 ====================

  def getRedisHost: String = config.getString("redis.host")

  def getRedisPort: Int = config.getInt("redis.port")

  def getRedisPassword: String = {
    sys.env.getOrElse("REDIS_PASSWORD", {
      if (config.hasPath("redis.password") && config.getString("redis.password") != null) {
        config.getString("redis.password")
      } else {
        "" // Redis 可以设置无密码
      }
    })
  }

  def getRedisDatabase: Int = Try(config.getInt("redis.database")).getOrElse(0)

  def getRedisTimeout: Int = Try(config.getInt("redis.timeout")).getOrElse(3000)

  def getRedisMaxTotal: Int = Try(config.getInt("redis.pool.max-total")).getOrElse(50)

  def getRedisMaxIdle: Int = Try(config.getInt("redis.pool.max-idle")).getOrElse(20)

  def getRedisMinIdle: Int = Try(config.getInt("redis.pool.min-idle")).getOrElse(5)

  def getRedisMaxWaitMillis: Long = Try(config.getLong("redis.pool.max-wait-millis")).getOrElse(3000L)

  def getRedisCacheTTL: Int = Try(config.getInt("redis.cache.ttl-seconds")).getOrElse(86400)

  def getRedisKeyPrefix: String = Try(config.getString("redis.cache.key-prefix")).getOrElse("user:profile:")

  // ==================== Kafka 配置 ====================

  def getKafkaBootstrapServers: String = config.getString("kafka.bootstrap-servers")

  def getKafkaGroupId: String = Try(config.getString("kafka.consumer.group-id")).getOrElse("user-profile-group")

  def getKafkaAutoOffsetReset: String = Try(config.getString("kafka.consumer.auto-offset-reset")).getOrElse("latest")

  def getKafkaMaxPollRecords: Int = Try(config.getInt("kafka.consumer.max-poll-records")).getOrElse(500)

  def getKafkaUserBehaviorTopic: String = Try(config.getString("kafka.topics.user-behavior")).getOrElse("user-behavior")

  def getKafkaAccessLogsTopic: String = Try(config.getString("kafka.topics.access-logs")).getOrElse("access-logs")

  def getKafkaProfileUpdateTopic: String = Try(config.getString("kafka.topics.profile-update")).getOrElse("profile-update")

  // ==================== Spark 配置 ====================

  def getSparkAppName: String = Try(config.getString("spark.app-name")).getOrElse("EcommerceUserProfile")

  def getSparkMaster: String = Try(config.getString("spark.master")).getOrElse("local[*]")

  def getSparkConfig: Map[String, String] = {
    Try {
      config.getConfig("spark.config").entrySet().asScala.map { entry =>
        entry.getKey -> entry.getValue.unwrapped().toString
      }.toMap
    }.getOrElse(Map.empty)
  }

  // ==================== HBase 配置 ====================

  def getHBaseZkQuorum: String = Try(config.getString("hbase.zookeeper-quorum")).getOrElse("localhost")

  def getHBaseZkClientPort: String = Try(config.getString("hbase.zookeeper-property-clientPort")).getOrElse("2181")

  def getHBaseUserBehaviorTable: String = Try(config.getString("hbase.tables.user-behavior")).getOrElse("user_behavior")

  def getHBaseUserTagsTable: String = Try(config.getString("hbase.tables.user-tags")).getOrElse("user_tags")

  // ==================== RFM 配置 ====================

  def getRFMPeriodDays: Int = Try(config.getInt("rfm.period-days")).getOrElse(365)

  def getRFMRThresholds: List[Int] = Try {
    config.getIntList("rfm.r-thresholds").asScala.map(_.intValue()).toList
  }.getOrElse(List(30, 60, 90, 180))

  def getRFMFThresholds: List[Int] = Try {
    config.getIntList("rfm.f-thresholds").asScala.map(_.intValue()).toList
  }.getOrElse(List(1, 5, 10, 20))

  def getRFMMThresholds: List[Int] = Try {
    config.getIntList("rfm.m-thresholds").asScala.map(_.intValue()).toList
  }.getOrElse(List(500, 2000, 5000, 10000))

  // ==================== 画像服务配置 ====================

  def getProfileServicePort: Int = Try(config.getInt("profile-service.port")).getOrElse(8080)

  def isDegradationEnabled: Boolean = Try(config.getBoolean("profile-service.degradation-enabled")).getOrElse(true)

  def getBatchMaxSize: Int = Try(config.getInt("profile-service.batch-max-size")).getOrElse(100)

  def getQueryTimeoutMs: Long = Try(config.getLong("profile-service.query-timeout-ms")).getOrElse(5000L)

  /**
   * 验证关键配置是否完整
   *
   * @return 验证结果，None表示通过，Some包含错误信息
   */
  def validate(): Option[String] = {
    val errors = scala.collection.mutable.ArrayBuffer[String]()

    if (getMySQLUrl.isEmpty) errors += "MySQL URL 未配置"
    if (getMySQLUser.isEmpty) errors += "MySQL 用户名未配置"
    if (getMySQLPassword.isEmpty) errors += "MySQL 密码未配置（请设置环境变量 MYSQL_PASSWORD）"
    if (getRedisHost.isEmpty) errors += "Redis 主机未配置"
    if (getKafkaBootstrapServers.isEmpty) errors += "Kafka bootstrap servers 未配置"

    if (errors.nonEmpty) {
      val msg = s"配置验证失败：\n${errors.mkString("\n")}"
      logger.error(msg)
      Some(msg)
    } else {
      logger.info("配置验证通过")
      None
    }
  }

  /**
   * 打印当前配置（隐藏敏感信息）
   */
  def printConfig(): Unit = {
    logger.info("========== 当前配置 ==========")
    logger.info(s"MySQL URL: $getMySQLUrl")
    logger.info(s"MySQL User: $getMySQLUser")
    logger.info(s"MySQL Password: ${if (getMySQLPassword.nonEmpty) "******" else "未配置"}")
    logger.info(s"Redis Host: $getRedisHost:$getRedisPort")
    logger.info(s"Redis Password: ${if (getRedisPassword.nonEmpty) "******" else "未配置"}")
    logger.info(s"Kafka Servers: $getKafkaBootstrapServers")
    logger.info(s"Spark Master: $getSparkMaster")
    logger.info(s"HBase ZK: $getHBaseZkQuorum:$getHBaseZkClientPort")
    logger.info(s"降级开关: $isDegradationEnabled")
    logger.info("==============================")
  }
}

package com.ecommerce.profile.dal

import com.ecommerce.profile.config.ConfigManager
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.slf4j.{Logger, LoggerFactory}

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}
import java.util.Properties
import scala.jdk.CollectionConverters._
import scala.util.{Try, Using}

/**
 * 数据访问层
 *
 * 封装所有数据源的读写操作，提供统一的数据访问接口。
 * 职责：
 * - MySQL 数据读写
 * - HBase 数据读写
 * - Redis 缓存操作
 * - Kafka 数据源接入
 *
 * 设计原则：
 * - 所有连接配置通过 ConfigManager 获取，不硬编码
 * - 提供连接池管理，避免频繁创建/销毁连接
 * - 统一异常处理，确保资源释放
 * - 读写操作支持重试机制
 */
object DataAccessLayer {

  private val logger: Logger = LoggerFactory.getLogger(DataAccessLayer.getClass)

  // ==================== MySQL 操作 ====================

  /**
   * 获取 MySQL JDBC 连接属性
   */
  private def getMySQLProperties: Properties = {
    val props = new Properties()
    props.setProperty("user", ConfigManager.getMySQLUser)
    props.setProperty("password", ConfigManager.getMySQLPassword)
    props.setProperty("driver", ConfigManager.getMySQLDriver)
    props.setProperty("useSSL", "true")
    props.setProperty("characterEncoding", "utf8mb4")
    props
  }

  /**
   * 从 MySQL 读取数据
   *
   * @param spark   SparkSession 实例
   * @param table   表名
   * @param columns 可选的列名列表，为空则读取全部列
   * @return DataFrame
   */
  def readFromMySQL(spark: SparkSession, table: String, columns: Seq[String] = Seq.empty): DataFrame = {
    require(table.nonEmpty, "表名不能为空")

    val query = if (columns.nonEmpty) {
      s"SELECT ${columns.mkString(", ")} FROM $table"
    } else {
      s"SELECT * FROM $table"
    }

    logger.info(s"从MySQL读取数据: $query")

    try {
      spark.read
        .format("jdbc")
        .option("url", ConfigManager.getMySQLUrl)
        .option("dbtable", s"($query) AS t")
        .option("user", ConfigManager.getMySQLUser)
        .option("password", ConfigManager.getMySQLPassword)
        .option("driver", ConfigManager.getMySQLDriver)
        .option("fetchsize", "1000")
        .option("numPartitions", "4")
        .option("partitionColumn", "user_id")
        .option("lowerBound", "1")
        .option("upperBound", "10000000")
        .load()
    } catch {
      case e: Exception =>
        logger.error(s"从MySQL读取数据失败: ${e.getMessage}", e)
        throw new RuntimeException(s"MySQL读取失败: ${e.getMessage}", e)
    }
  }

  /**
   * 写入数据到 MySQL
   *
   * @param df       待写入的 DataFrame
   * @param table    目标表名
   * @param saveMode 写入模式（overwrite/append/upsert）
   */
  def writeToMySQL(df: DataFrame, table: String, saveMode: String = "overwrite"): Unit = {
    require(table.nonEmpty, "表名不能为空")

    logger.info(s"写入数据到MySQL表: $table, 模式: $saveMode, 记录数: ${df.count()}")

    try {
      df.write
        .format("jdbc")
        .option("url", ConfigManager.getMySQLUrl)
        .option("dbtable", table)
        .option("user", ConfigManager.getMySQLUser)
        .option("password", ConfigManager.getMySQLPassword)
        .option("driver", ConfigManager.getMySQLDriver)
        .option("batchsize", "1000")
        .option("isolationLevel", "READ_COMMITTED")
        .mode(saveMode)
        .save()

      logger.info(s"数据写入MySQL成功: $table")
    } catch {
      case e: Exception =>
        logger.error(s"写入MySQL失败: ${e.getMessage}", e)
        throw new RuntimeException(s"MySQL写入失败: ${e.getMessage}", e)
    }
  }

  /**
   * 使用 JDBC 直连执行 SQL 查询（适用于小数据量查询）
   *
   * @param sql    SQL 语句
   * @param handler 结果集处理函数
   * @tparam T 返回类型
   * @return 查询结果
   */
  def executeQuery[T](sql: String)(handler: ResultSet => T): Option[T] = {
    var connection: Connection = null
    var stmt: PreparedStatement = null
    try {
      connection = DriverManager.getConnection(
        ConfigManager.getMySQLUrl,
        ConfigManager.getMySQLUser,
        ConfigManager.getMySQLPassword
      )
      stmt = connection.prepareStatement(sql)
      val rs = stmt.executeQuery()
      Some(handler(rs))
    } catch {
      case e: Exception =>
        logger.error(s"执行SQL查询失败: ${e.getMessage}", e)
        None
    } finally {
      if (stmt != null) Try(stmt.close())
      if (connection != null) Try(connection.close())
    }
  }

  /**
   * 使用 JDBC 直连执行 SQL 更新
   *
   * @param sql SQL 语句
   * @return 受影响的行数
   */
  def executeUpdate(sql: String): Int = {
    var connection: Connection = null
    var stmt: PreparedStatement = null
    try {
      connection = DriverManager.getConnection(
        ConfigManager.getMySQLUrl,
        ConfigManager.getMySQLUser,
        ConfigManager.getMySQLPassword
      )
      stmt = connection.prepareStatement(sql)
      stmt.executeUpdate()
    } catch {
      case e: Exception =>
        logger.error(s"执行SQL更新失败: ${e.getMessage}", e)
        -1
    } finally {
      if (stmt != null) Try(stmt.close())
      if (connection != null) Try(connection.close())
    }
  }

  // ==================== Kafka 操作 ====================

  /**
   * 从 Kafka 读取流数据
   *
   * @param spark  SparkSession 实例
   * @param topic  Kafka 主题
   * @return DataFrame
   */
  def readFromKafka(spark: SparkSession, topic: String): DataFrame = {
    require(topic.nonEmpty, "Kafka主题不能为空")

    logger.info(s"从Kafka读取流数据: $topic")

    try {
      spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", ConfigManager.getKafkaBootstrapServers)
        .option("subscribe", topic)
        .option("startingOffsets", ConfigManager.getKafkaAutoOffsetReset)
        .option("maxOffsetsPerTrigger", ConfigManager.getKafkaMaxPollRecords.toString)
        .option("failOnDataLoss", "false")
        .load()
    } catch {
      case e: Exception =>
        logger.error(s"从Kafka读取数据失败: ${e.getMessage}", e)
        throw new RuntimeException(s"Kafka读取失败: ${e.getMessage}", e)
    }
  }

  /**
   * 从 Kafka 批量读取数据
   *
   * @param spark    SparkSession 实例
   * @param topic    Kafka 主题
   * @param startingOffsets 起始偏移量（默认 earliest）
   * @param endingOffsets   结束偏移量（默认 latest）
   * @return DataFrame
   */
  def readBatchFromKafka(spark: SparkSession, topic: String,
                         startingOffsets: String = "earliest",
                         endingOffsets: String = "latest"): DataFrame = {
    require(topic.nonEmpty, "Kafka主题不能为空")

    logger.info(s"从Kafka批量读取数据: $topic")

    try {
      spark.read
        .format("kafka")
        .option("kafka.bootstrap.servers", ConfigManager.getKafkaBootstrapServers)
        .option("subscribe", topic)
        .option("startingOffsets", startingOffsets)
        .option("endingOffsets", endingOffsets)
        .load()
    } catch {
      case e: Exception =>
        logger.error(s"从Kafka批量读取数据失败: ${e.getMessage}", e)
        throw new RuntimeException(s"Kafka批量读取失败: ${e.getMessage}", e)
    }
  }

  // ==================== HBase 操作 ====================

  /**
   * 获取 HBase 连接配置
   */
  def getHBaseConfig: org.apache.hadoop.conf.Configuration = {
    val conf = org.apache.hadoop.hbase.HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", ConfigManager.getHBaseZkQuorum)
    conf.set("hbase.zookeeper.property.clientPort", ConfigManager.getHBaseZkClientPort)
    conf
  }

  // ==================== SparkSession 管理 ====================

  /**
   * 创建 SparkSession
   *
   * 使用 try-catch-finally 确保资源释放。
   * 所有 SparkSession 的创建都应通过此方法，以统一配置和资源管理。
   *
   * @param appName   应用名称
   * @param hiveSupport 是否启用Hive支持
   * @return SparkSession 实例
   */
  def createSparkSession(appName: String = ConfigManager.getSparkAppName,
                         hiveSupport: Boolean = false): SparkSession = {
    logger.info(s"创建SparkSession: $appName, Hive支持: $hiveSupport")

    try {
      val builder = SparkSession.builder()
        .appName(appName)
        .master(ConfigManager.getSparkMaster)

      // 应用 Spark 配置
      ConfigManager.getSparkConfig.foreach { case (key, value) =>
        builder.config(key, value)
      }

      if (hiveSupport) {
        builder.enableHiveSupport()
      }

      val spark = builder.getOrCreate()
      spark.sparkContext.setLogLevel("WARN")
      logger.info(s"SparkSession创建成功: $appName")
      spark
    } catch {
      case e: Exception =>
        logger.error(s"创建SparkSession失败: ${e.getMessage}", e)
        throw new RuntimeException(s"SparkSession创建失败: ${e.getMessage}", e)
    }
  }

  /**
   * 安全执行 Spark 作业
   *
   * 使用 try-catch-finally 确保 SparkSession 正确释放。
   * 这是执行 Spark 作业的推荐方式。
   *
   * @param appName   应用名称
   * @param hiveSupport 是否启用Hive支持
   * @param block     业务逻辑块
   * @tparam T 返回类型
   * @return 业务逻辑执行结果
   */
  def withSparkSession[T](appName: String = ConfigManager.getSparkAppName,
                           hiveSupport: Boolean = false)(block: SparkSession => T): T = {
    try {
      val spark = createSparkSession(appName, hiveSupport)
      block(spark)
    } catch {
      case e: Exception =>
        logger.error(s"Spark作业执行失败: ${e.getMessage}", e)
        throw new RuntimeException(s"Spark作业执行失败: ${e.getMessage}", e)
    } finally {
      val s = SparkSession.getActiveSession.orNull
      if (s != null) {
        try {
          s.stop()
          logger.info(s"SparkSession已释放: $appName")
        } catch {
          case e: Exception =>
            logger.warn(s"释放SparkSession时出错: ${e.getMessage}", e)
        }
      }
    }
  }
}

package com.ecommerce.profile.service

import com.ecommerce.profile.config.ConfigManager
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Get}
import org.apache.hadoop.hbase.util.Bytes
import org.slf4j.{Logger, LoggerFactory}
import redis.clients.jedis.{Jedis, JedisPool, JedisPoolConfig}

import java.util
import scala.jdk.CollectionConverters._
import scala.util.{Try, Using}

/**
 * 用户画像服务
 *
 * 提供用户画像的查询、缓存和降级功能。
 *
 * 核心特性：
 * - Redis 连接池管理（JedisPool）
 * - 多级缓存策略（Redis -> HBase -> MySQL）
 * - 降级方案（Redis不可用时自动降级到HBase）
 * - 数据验证和异常处理
 *
 * 缓存策略：
 * 1. 首先查询 Redis 热数据缓存
 * 2. Redis 未命中时，查询 HBase 标签数据
 * 3. HBase 未命中时，查询 MySQL 结构化数据
 * 4. 查询结果回填到 Redis 缓存
 *
 * 降级方案：
 * - Redis 不可用 -> 直接查询 HBase
 * - HBase 不可用 -> 查询 MySQL（功能受限）
 * - 全部不可用 -> 返回默认画像数据
 */
object UserProfileService {

  private val logger: Logger = LoggerFactory.getLogger(UserProfileService.getClass)

  // Redis 连接池（懒加载，线程安全）
  private lazy val jedisPool: JedisPool = {
    val poolConfig = new JedisPoolConfig()
    poolConfig.setMaxTotal(ConfigManager.getRedisMaxTotal)
    poolConfig.setMaxIdle(ConfigManager.getRedisMaxIdle)
    poolConfig.setMinIdle(ConfigManager.getRedisMinIdle)
    poolConfig.setMaxWaitMillis(ConfigManager.getRedisMaxWaitMillis)
    poolConfig.setTestOnBorrow(true)
    poolConfig.setTestOnReturn(false)
    poolConfig.setTestWhileIdle(true)

    val password = ConfigManager.getRedisPassword
    if (password.nonEmpty) {
      new JedisPool(poolConfig, ConfigManager.getRedisHost, ConfigManager.getRedisPort,
        ConfigManager.getRedisTimeout, password)
    } else {
      new JedisPool(poolConfig, ConfigManager.getRedisHost, ConfigManager.getRedisPort,
        ConfigManager.getRedisTimeout)
    }
  }

  // HBase 连接（懒加载）
  @volatile private var hbaseConnection: Connection = _

  private def getHBaseConnection: Connection = {
    if (hbaseConnection == null || hbaseConnection.isClosed) {
      synchronized {
        if (hbaseConnection == null || hbaseConnection.isClosed) {
          val conf = HBaseConfiguration.create()
          conf.set("hbase.zookeeper.quorum", ConfigManager.getHBaseZkQuorum)
          conf.set("hbase.zookeeper.property.clientPort", ConfigManager.getHBaseZkClientPort)
          hbaseConnection = ConnectionFactory.createConnection(conf)
          logger.info("HBase连接已创建")
        }
      }
    }
    hbaseConnection
  }

  // ==================== Redis 操作 ====================

  /**
   * 从 Redis 获取资源
   *
   * 使用 JedisPool 连接池管理 Redis 连接，确保连接正确归还。
   *
   * @param block 使用 Jedis 执行的操作
   * @tparam T 返回类型
   * @return 操作结果，Redis不可用时返回None
   */
  private def withRedis[T](block: Jedis => T): Option[T] = {
    var jedis: Jedis = null
    try {
      jedis = jedisPool.getResource
      Some(block(jedis))
    } catch {
      case e: Exception =>
        logger.warn(s"Redis操作失败，将降级处理: ${e.getMessage}")
        None
    } finally {
      if (jedis != null) {
        try {
          jedis.close()
        } catch {
          case e: Exception =>
            logger.warn(s"归还Redis连接失败: ${e.getMessage}")
        }
      }
    }
  }

  /**
   * 从 Redis 缓存获取用户画像
   *
   * @param userId 用户ID
   * @return 画像JSON字符串，未命中返回None
   */
  def getProfileFromCache(userId: Int): Option[String] = {
    val key = s"${ConfigManager.getRedisKeyPrefix}$userId"
    withRedis { jedis =>
      val value = jedis.get(key)
      if (value != null) {
        logger.debug(s"Redis缓存命中: userId=$userId")
        value
      } else {
        logger.debug(s"Redis缓存未命中: userId=$userId")
        null
      }
    }.flatMap(Option(_))
  }

  /**
   * 将用户画像写入 Redis 缓存
   *
   * @param userId    用户ID
   * @param profileJson 画像JSON字符串
   */
  def setProfileToCache(userId: Int, profileJson: String): Unit = {
    val key = s"${ConfigManager.getRedisKeyPrefix}$userId"
    withRedis { jedis =>
      jedis.setex(key, ConfigManager.getRedisCacheTTL, profileJson)
      logger.debug(s"用户画像已写入Redis缓存: userId=$userId, TTL=${ConfigManager.getRedisCacheTTL}s")
    }
  }

  /**
   * 批量将用户画像写入 Redis 缓存
   *
   * 使用 Pipeline 提升批量写入性能。
   *
   * @param profiles 用户ID到画像JSON的映射
   */
  def batchSetProfileToCache(profiles: Map[Int, String]): Unit = {
    if (profiles.isEmpty) return

    withRedis { jedis =>
      val pipeline = jedis.pipelined()
      profiles.foreach { case (userId, profileJson) =>
        val key = s"${ConfigManager.getRedisKeyPrefix}$userId"
        pipeline.setex(key, ConfigManager.getRedisCacheTTL, profileJson)
      }
      pipeline.sync()
      logger.info(s"批量写入Redis缓存完成: ${profiles.size}条记录")
    }
  }

  /**
   * 从 Redis 缓存删除用户画像
   *
   * @param userId 用户ID
   */
  def removeProfileFromCache(userId: Int): Unit = {
    val key = s"${ConfigManager.getRedisKeyPrefix}$userId"
    withRedis { jedis =>
      jedis.del(key)
      logger.debug(s"用户画像已从Redis缓存删除: userId=$userId")
    }
  }

  // ==================== HBase 操作 ====================

  /**
   * 从 HBase 获取用户标签数据
   *
   * @param userId 用户ID
   * @return 标签数据Map，查询失败返回None
   */
  def getTagsFromHBase(userId: Int): Option[Map[String, String]] = {
    try {
      val connection = getHBaseConnection
      val table = connection.getTable(TableName.valueOf(ConfigManager.getHBaseUserTagsTable))
      try {
        val get = new Get(Bytes.toBytes(userId.toString))
        val result = table.get(get)

        if (result.isEmpty) {
          logger.debug(s"HBase中未找到用户标签: userId=$userId")
          None
        } else {
          val tags = result.getNoVersionMap.asScala.flatMap { case (cf, columns) =>
            columns.asScala.map { case (qualifier, value) =>
              Bytes.toString(qualifier) -> Bytes.toString(value)
            }
          }.toMap
          logger.debug(s"HBase标签查询成功: userId=$userId, 标签数=${tags.size}")
          Some(tags)
        }
      } finally {
        table.close()
      }
    } catch {
      case e: Exception =>
        logger.warn(s"HBase查询失败，将降级处理: userId=$userId, error=${e.getMessage}")
        None
    }
  }

  // ==================== 多级查询（核心方法） ====================

  /**
   * 获取用户画像（多级缓存查询）
   *
   * 查询顺序：Redis -> HBase -> MySQL
   * 每级查询成功后，会回填到上一级缓存。
   *
   * @param userId 用户ID
   * @return 用户画像JSON字符串
   */
  def getUserProfile(userId: Int): Option[String] = {
    require(userId > 0, s"用户ID必须大于0，当前值: $userId")

    logger.info(s"查询用户画像: userId=$userId")

    // 第一级：查询 Redis 缓存
    val redisResult = getProfileFromCache(userId)
    if (redisResult.isDefined) {
      logger.info(s"用户画像查询成功（Redis缓存）: userId=$userId")
      return redisResult
    }

    // 第二级：查询 HBase
    val hbaseResult = getTagsFromHBase(userId)
    if (hbaseResult.isDefined) {
      val profileJson = hbaseResult.get.map { case (k, v) => s""""$k":"$v"""" }.mkString("{", ",", "}")
      // 回填 Redis 缓存
      setProfileToCache(userId, profileJson)
      logger.info(s"用户画像查询成功（HBase，已回填Redis）: userId=$userId")
      return Some(profileJson)
    }

    // 第三级：查询 MySQL（降级方案）
    if (ConfigManager.isDegradationEnabled) {
      val mysqlResult = getProfileFromMySQL(userId)
      if (mysqlResult.isDefined) {
        // 回填 Redis 和 HBase 缓存
        setProfileToCache(userId, mysqlResult.get)
        logger.info(s"用户画像查询成功（MySQL降级，已回填缓存）: userId=$userId")
        return mysqlResult
      }
    }

    logger.warn(s"用户画像未找到: userId=$userId")
    None
  }

  /**
   * 批量获取用户画像
   *
   * @param userIds 用户ID列表
   * @return 用户ID到画像的映射
   */
  def batchGetUserProfile(userIds: Seq[Int]): Map[Int, String] = {
    require(userIds.nonEmpty, "用户ID列表不能为空")
    require(userIds.size <= ConfigManager.getBatchMaxSize,
      s"批量查询数量不能超过${ConfigManager.getBatchMaxSize}，当前: ${userIds.size}")

    logger.info(s"批量查询用户画像: count=${userIds.size}")

    val result = scala.collection.mutable.Map[Int, String]()

    // 先从 Redis 批量查询
    val missedIds = scala.collection.mutable.ArrayBuffer[Int]()
    withRedis { jedis =>
      val pipeline = jedis.pipelined()
      val keys = userIds.map(id => s"${ConfigManager.getRedisKeyPrefix}$id")
      keys.foreach(key => pipeline.get(key))
      val responses = pipeline.syncAndReturnAll().asScala.zipWithIndex

      responses.foreach { case (response, index) =>
        val value = response.asInstanceOf[String]
        if (value != null) {
          result(userIds(index)) = value
        } else {
          missedIds += userIds(index)
        }
      }
    }

    // Redis 未命中的从 HBase 查询
    if (missedIds.nonEmpty) {
      val hbaseProfiles = scala.collection.mutable.Map[Int, String]()
      missedIds.foreach { userId =>
        val hbaseResult = getTagsFromHBase(userId)
        if (hbaseResult.isDefined) {
          val profileJson = hbaseResult.get.map { case (k, v) => s""""$k":"$v"""" }.mkString("{", ",", "}")
          hbaseProfiles(userId) = profileJson
          result(userId) = profileJson
        }
      }

      // 回填 Redis 缓存
      if (hbaseProfiles.nonEmpty) {
        batchSetProfileToCache(hbaseProfiles.toMap)
      }
    }

    logger.info(s"批量查询完成: 总查询${userIds.size}条, Redis命中${userIds.size - missedIds.size}条, HBase命中${result.size - (userIds.size - missedIds.size)}条")
    result.toMap
  }

  /**
   * 从 MySQL 查询用户画像（降级方案）
   *
   * @param userId 用户ID
   * @return 画像JSON字符串
   */
  private def getProfileFromMySQL(userId: Int): Option[String] = {
    import com.ecommerce.profile.dal.DataAccessLayer._

    executeQuery(
      s"SELECT user_id, age_group, gender, purchase_power, shopping_frequency, " +
        s"rfm_score, user_segment FROM user_profiles WHERE user_id = $userId"
    ) { rs =>
      if (rs.next()) {
        val profile = Map(
          "user_id" -> rs.getInt("user_id").toString,
          "age_group" -> Option(rs.getString("age_group")).getOrElse("未知"),
          "gender" -> Option(rs.getString("gender")).getOrElse("未知"),
          "purchase_power" -> Option(rs.getString("purchase_power")).getOrElse("未知"),
          "shopping_frequency" -> Option(rs.getString("shopping_frequency")).getOrElse("未知"),
          "rfm_score" -> Option(rs.getString("rfm_score")).getOrElse("000"),
          "user_segment" -> Option(rs.getString("user_segment")).getOrElse("未分类")
        )
        profile.map { case (k, v) => s""""$k":"$v"""" }.mkString("{", ",", "}")
      } else {
        null
      }
    }.flatMap(Option(_))
  }

  // ==================== 资源管理 ====================

  /**
   * 关闭所有连接资源
   *
   * 在应用关闭时调用，确保所有连接正确释放。
   */
  def shutdown(): Unit = {
    logger.info("正在关闭用户画像服务资源...")

    // 关闭 Redis 连接池
    try {
      if (jedisPool != null && !jedisPool.isClosed) {
        jedisPool.close()
        logger.info("Redis连接池已关闭")
      }
    } catch {
      case e: Exception =>
        logger.warn(s"关闭Redis连接池时出错: ${e.getMessage}")
    }

    // 关闭 HBase 连接
    try {
      if (hbaseConnection != null && !hbaseConnection.isClosed) {
        hbaseConnection.close()
        logger.info("HBase连接已关闭")
      }
    } catch {
      case e: Exception =>
        logger.warn(s"关闭HBase连接时出错: ${e.getMessage}")
    }

    logger.info("用户画像服务资源关闭完成")
  }

  // 注册 JVM 关闭钩子，确保资源释放
  sys.addShutdownHook {
    shutdown()
  }
}