+------------------------------------------------------------------+
|                        Lambda 架构                                |
+------------------------------------------------------------------+
|                                                                    |
|  +----------------+     +------------------+     +--------------+ |
|  |   批处理层     |     |   速度层         |     |   服务层     | |
|  | (Batch Layer)  |     | (Speed Layer)    |     | (Serving     | |
|  |                |     |                  |     |  Layer)      | |
|  | - HDFS原始数据 |     | - Spark Streaming|     | - 查询合并   | |
|  | - Spark批处理  |     | - 实时计算       |     | - MySQL      | |
|  | - 全量重算     |     | - 增量计算       |     | - HBase      | |
|  +-------+--------+     +--------+---------+     +------+-------+ |
|          |                       |                       |        |
|          +-------+------+--------+                       |        |
|                  |      |                                |        |
|          +-------v------v--------+                       |        |
|          |      数据输入层       +-----------------------+        |
|          |   (Kafka消息队列)     |                                |
|          +----------------------+                                 |
+------------------------------------------------------------------+

+------------------------------------------------------------------+
|                        Kappa 架构                                 |
+------------------------------------------------------------------+
|                                                                    |
|  +------------------+     +------------------+                    |
|  |   统一处理层     |     |   服务层         |                    |
|  | (Stream Layer)   |     | (Serving Layer)  |                    |
|  |                  |     |                  |                    |
|  | - Kafka保留日志  |     | - 实时查询       |                    |
|  | - Spark Streaming|     | - MySQL          |                    |
|  | - 流批一体       |     | - HBase          |                    |
|  +--------+---------+     +--------+---------+                    |
|           |                        |                              |
|           +----------+-------------+                              |
|                      |                                            |
|           +----------v------------+                               |
|           |    数据输入层         |                               |
|           |  (Kafka消息队列)      |                               |
|           +----------------------+                                |
+------------------------------------------------------------------+

时间轴:  t0        t1        t2        t3        t4
         |         |         |         |         |
数据流:  =====>    =====>    =====>    =====>    =====>
         |         |         |         |         |
切分:   [批次0]   [批次1]   [批次2]   [批次3]   [批次4]
         |         |         |         |         |
         v         v         v         v         v
       RDD-0     RDD-1     RDD-2     RDD-3     RDD-4

时间轴:  t0   t1   t2   t3   t4   t5   t6   t7
数据:    [A]  [B]  [C]  [D]  [E]  [F]  [G]  [H]

窗口长度=3个批次, 滑动步长=2个批次:

窗口1:  [A]  [B]  [C]                    -> 计算结果1
              窗口2:  [C]  [D]  [E]       -> 计算结果2
                         窗口3:  [E]  [F]  [G]  -> 计算结果3

生产者 (Kafka)                    消费者 (Spark Streaming)
   |                                  |
   |  数据速率: 10000条/秒            |  处理速率: 5000条/秒
   |  --------->                      |  <---------
   |                                  |
   |        背压控制器                 |
   |   +----------------------+       |
   |   | PID控制器算法:       |       |
   |   | 1. 采集处理速率      |       |
   |   | 2. 计算速率差值      |       |
   |   | 3. 动态调整消费速率  |       |
   |   +----------------------+       |
   |                                  |
   |  调整后: 5000条/秒               |  处理速率: 5000条/秒
   |  --------->                      |  <---------

+------------------+     +------------------+     +------------------+     +------------------+
|   数据采集层     | --> |   数据存储层     | --> |   数据处理层     | --> |   可视化层       |
+------------------+     +------------------+     +------------------+     +------------------+
| - Kafka 3.7.0    |     | - HDFS (Hadoop)  |     | - Spark Core     |     | - Grafana        |
| - Flume          |     | - HBase 2.5.9    |     | - Spark SQL      |     | - Tableau        |
| - OBD数据网关    |     | - MySQL 8.0      |     | - Spark Streaming|     | - 自研Dashboard  |
+------------------+     +------------------+     | - Hive 3.1.3     |     +------------------+
                                                  +------------------+
                                                          |
                                                  +------------------+
                                                  |   基础设施层     |
                                                  +------------------+
                                                  | - ConfigManager  |
                                                  | - DataAccessLayer|
                                                  | - 降级策略       |
                                                  +------------------+

<?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.sparklearning</groupId>
    <artifactId>auto-bigdata-analysis</artifactId>
    <version>1.0.0</version>
    <packaging>jar</packaging>

    <name>Auto BigData Analysis System</name>
    <description>汽车大数据分析系统 - 基于Spark的企业级车联网数据处理平台</description>

    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <java.version>11</java.version>
        <scala.version>2.13.8</scala.version>
        <scala.binary.version>2.13</scala.binary.version>
        <spark.version>3.5.8</spark.version>
        <kafka.version>3.7.0</kafka.version>
        <hbase.version>2.5.9</hbase.version>
        <hadoop.version>3.3.6</hadoop.version>
        <hive.version>3.1.3</hive.version>
        <mysql.connector.version>8.0.33</mysql.connector.version>
        <typesafe.config.version>1.4.3</typesafe.config.version>
        <slf4j.version>1.7.36</slf4j.version>
        <log4j.version>2.17.2</log4j.version>
    </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 Connector -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-streaming-kafka-0-10_${scala.binary.version}</artifactId>
            <version>${spark.version}</version>
        </dependency>

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

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

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

        <!-- HBase MapReduce (用于Spark与HBase集成) -->
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-mapreduce</artifactId>
            <version>${hbase.version}</version>
        </dependency>

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

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

        <!-- Hive JDBC -->
        <dependency>
            <groupId>org.apache.hive</groupId>
            <artifactId>hive-jdbc</artifactId>
            <version>${hive.version}</version>
            <scope>provided</scope>
        </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>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.logging.log4j</groupId>
            <artifactId>log4j-slf4j-impl</artifactId>
            <version>${log4j.version}</version>
            <scope>runtime</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>
                        <id>scala-compile-first</id>
                        <phase>process-resources</phase>
                        <goals>
                            <goal>add-source</goal>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                    <execution>
                        <id>scala-test-compile</id>
                        <phase>process-test-resources</phase>
                        <goals>
                            <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>
                    <jvmArgs>
                        <jvmArg>-Xms128m</jvmArg>
                        <jvmArg>-Xmx1024m</jvmArg>
                    </jvmArgs>
                </configuration>
            </plugin>

            <!-- Maven Compiler Plugin -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.11.0</version>
                <configuration>
                    <source>${java.version}</source>
                    <target>${java.version}</target>
                    <encoding>UTF-8</encoding>
                </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>
        </plugins>
    </build>
</project>

package com.sparklearning.auto.config

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

import scala.jdk.CollectionConverters._
import scala.util.{Try, Success, Failure}

/**
 * 配置管理器 - 统一管理应用配置
 *
 * 设计原则：
 * 1. 所有配置从application.conf读取，禁止硬编码
 * 2. 敏感信息（密码等）通过环境变量覆盖
 * 3. 提供类型安全的配置访问方法
 * 4. 配置加载失败时提供有意义的错误信息
 *
 * 使用方式：
 *   val config = ConfigManager.getConfig
 *   val mysqlUrl = ConfigManager.getMySQLUrl
 */
object ConfigManager {

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

  // 加载配置，优先加载环境变量覆盖
  private lazy val config: Config = {
    val baseConfig = ConfigFactory.load()
    val envConfig = ConfigFactory.systemEnvironment()
    // 环境变量优先级最高，然后是系统属性，最后是配置文件
    ConfigFactory.load(
      ConfigFactory.systemProperties()
        .withFallback(envConfig)
        .withFallback(baseConfig)
    ).resolve()
  }

  /**
   * 获取完整配置对象
   */
  def getConfig: Config = config

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

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

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

  def getMySQLPassword: String = {
    val password = config.getString("mysql.password")
    if (password.contains("请通过环境变量") || password.isEmpty) {
      logger.warn("MySQL密码未通过环境变量配置，请在MYSQL_PASSWORD环境变量中设置密码")
    }
    password
  }

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

  /**
   * 获取MySQL连接选项（用于Spark JDBC）
   */
  def getMySQLOptions(table: String): Map[String, String] = {
    require(table.nonEmpty, "表名不能为空")
    Map(
      "url" -> getMySQLUrl,
      "dbtable" -> table,
      "user" -> getMySQLUser,
      "password" -> getMySQLPassword,
      "driver" -> getMySQLDriver
    )
  }

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

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

  def getKafkaConsumerGroupId: String = config.getString("kafka.consumer.group-id")

  def getKafkaConsumerConfig: Map[String, String] = Map(
    "bootstrap.servers" -> getKafkaBootstrapServers,
    "group.id" -> getKafkaConsumerGroupId,
    "auto.offset.reset" -> config.getString("kafka.consumer.auto-offset-reset"),
    "enable.auto.commit" -> config.getString("kafka.consumer.enable-auto-commit"),
    "max.poll.records" -> config.getString("kafka.consumer.max-poll-records"),
    "session.timeout.ms" -> config.getString("kafka.consumer.session-timeout-ms"),
    "heartbeat.interval.ms" -> config.getString("kafka.consumer.heartbeat-interval-ms")
  )

  def getKafkaProducerConfig: java.util.Properties = {
    val props = new java.util.Properties()
    props.put("bootstrap.servers", getKafkaBootstrapServers)
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("acks", config.getString("kafka.producer.acks"))
    props.put("retries", config.getString("kafka.producer.retries"))
    props.put("batch.size", config.getString("kafka.producer.batch-size"))
    props.put("linger.ms", config.getString("kafka.producer.linger-ms"))
    props.put("buffer.memory", config.getString("kafka.producer.buffer-memory"))
    props
  }

  def getKafkaTopic(topicKey: String): String = {
    Try(config.getString(s"kafka.topics.$topicKey")) match {
      case Success(topic) => topic
      case Failure(_) =>
        logger.warn(s"未找到Kafka主题配置: kafka.topics.$topicKey，使用默认值: $topicKey")
        topicKey
    }
  }

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

  def getHBaseZookeeperQuorum: String = config.getString("hbase.zookeeper-quorum")

  def getHBaseZookeeperClientPort: String = config.getString("hbase.zookeeper-client-port")

  def getHBaseTableName(tableKey: String): String = {
    Try(config.getString(s"hbase.tables.$tableKey")) match {
      case Success(table) => table
      case Failure(_) =>
        logger.warn(s"未找到HBase表配置: hbase.tables.$tableKey，使用默认值: $tableKey")
        tableKey
    }
  }

  /**
   * 获取HBase Configuration对象
   */
  def getHBaseConfig: org.apache.hadoop.conf.Configuration = {
    val conf = org.apache.hadoop.hbase.HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", getHBaseZookeeperQuorum)
    conf.set("hbase.zookeeper.property.clientPort", getHBaseZookeeperClientPort)
    conf.set("hbase.rpc.timeout", config.getString("hbase.rpc-timeout").toString)
    conf.set("hbase.client.retries.number", config.getString("hbase.client-retries-number").toString)
    conf.set("hbase.client.operation.timeout", config.getString("hbase.client-operation-timeout").toString)
    conf
  }

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

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

  def getSparkMaster: String = config.getString("spark.master")

  def getSparkBatchDurationSeconds: Int = config.getInt("spark.streaming.batch-duration-seconds")

  def getSparkCheckpointDir: String = config.getString("spark.streaming.checkpoint-dir")

  def getSparkShufflePartitions: Int = config.getInt("spark.sql.shuffle-partitions")

  // ========================================
  // 数据验证配置
  // ========================================

  /**
   * 获取传感器值范围
   * @param sensorType 传感器类型
   * @return (最小值, 最大值)
   */
  def getSensorRange(sensorType: String): (Double, Double) = {
    val path = s"validation.sensor-ranges.$sensorType"
    Try {
      val min = config.getDouble(s"$path.min")
      val max = config.getDouble(s"$path.max")
      (min, max)
    } match {
      case Success(range) => range
      case Failure(_) =>
        logger.warn(s"未找到传感器范围配置: $path，使用默认范围(0.0, 100.0)")
        (0.0, 100.0)
    }
  }

  def getSalePriceRange: (Double, Double) = {
    (config.getDouble("validation.sale-price.min"), config.getDouble("validation.sale-price.max"))
  }

  def getCustomerAgeRange: (Int, Int) = {
    (config.getInt("validation.customer-age.min"), config.getInt("validation.customer-age.max"))
  }

  // ========================================
  // 降级策略配置
  // ========================================

  def isDegradationEnabled: Boolean = config.getBoolean("degradation.enabled")

  def getMaxRetries: Int = config.getInt("degradation.max-retries")

  def getRetryIntervalMs: Long = config.getLong("degradation.retry-interval-ms")

  def getBatchFallbackThreshold: Int = config.getInt("degradation.batch-fallback-threshold")

  def getFallbackDataPath: String = config.getString("degradation.fallback-data-path")

  // ========================================
  // 数据库表名配置
  // ========================================

  def getDatabaseTable(tableKey: String): String = {
    Try(config.getString(s"database.tables.$tableKey")) match {
      case Success(table) => table
      case Failure(_) =>
        logger.warn(s"未找到数据库表配置: database.tables.$tableKey，使用默认值: $tableKey")
        tableKey
    }
  }
}

package com.sparklearning.auto.dataaccess

import com.sparklearning.auto.config.ConfigManager
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client.{Connection, ConnectionFactory, Put, Scan, Table}
import org.apache.hadoop.hbase.util.Bytes
import org.slf4j.{Logger, LoggerFactory}

import java.sql.{Connection => SQLConnection, DriverManager, PreparedStatement, ResultSet}
import scala.jdk.CollectionConverters._
import scala.util.{Try, Success, Failure, Using}

/**
 * 数据访问层 - 统一封装数据库操作
 *
 * 设计原则：
 * 1. 所有数据库连接通过ConfigManager获取配置
 * 2. 使用loan pattern确保资源释放
 * 3. 提供统一的异常处理和日志记录
 * 4. 连接池管理（简化实现，生产环境建议使用HikariCP）
 */
object DataAccessLayer {

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

  // ========================================
  // MySQL 数据访问
  // ========================================

  /**
   * 获取MySQL连接
   * 使用loan pattern，确保连接在使用后自动关闭
   *
   * @param operation 数据库操作函数
   * @tparam T 返回值类型
   * @return 操作结果
   */
  def withMySQLConnection[T](operation: SQLConnection => T): Try[T] = {
    var connection: SQLConnection = null
    try {
      Class.forName(ConfigManager.getMySQLDriver)
      connection = DriverManager.getConnection(
        ConfigManager.getMySQLUrl,
        ConfigManager.getMySQLUser,
        ConfigManager.getMySQLPassword
      )
      connection.setAutoCommit(false)
      val result = operation(connection)
      connection.commit()
      Success(result)
    } catch {
      case e: Exception =>
        if (connection != null) {
          try {
            connection.rollback()
          } catch {
            case rollbackEx: Exception =>
              logger.error("MySQL回滚失败", rollbackEx)
          }
        }
        logger.error("MySQL操作失败", e)
        Failure(e)
    } finally {
      if (connection != null) {
        try {
          connection.close()
        } catch {
          case e: Exception =>
            logger.error("MySQL连接关闭失败", e)
        }
      }
    }
  }

  /**
   * 批量写入MySQL
   *
   * @param tableKey 表配置键名
   * @param columns 列名列表
   * @param records 数据记录列表
   * @return 成功写入的记录数
   */
  def batchInsertToMySQL(tableKey: String, columns: Seq[String],
                         records: Seq[Seq[Any]]): Try[Int] = {
    withMySQLConnection { conn =>
      val tableName = ConfigManager.getDatabaseTable(tableKey)
      val placeholders = columns.map(_ => "?").mkString(",")
      val sql = s"INSERT INTO $tableName (${columns.mkString(",")}) VALUES ($placeholders)"

      var successCount = 0
      val stmt = conn.prepareStatement(sql)
      try {
        records.foreach { record =>
          record.zipWithIndex.foreach { case (value, idx) =>
            stmt.setObject(idx + 1, value)
          }
          stmt.addBatch()
          successCount += 1
        }
        stmt.executeBatch()
        successCount
      } finally {
        stmt.close()
      }
    }
  }

  /**
   * 执行查询并处理结果
   *
   * @param sql SQL查询语句
   * @param processor 结果处理函数
   * @tparam T 返回值类型
   * @return 查询结果
   */
  def queryMySQL[T](sql: String)(processor: ResultSet => T): Try[T] = {
    withMySQLConnection { conn =>
      val stmt = conn.createStatement()
      var rs: ResultSet = null
      try {
        rs = stmt.executeQuery(sql)
        processor(rs)
      } finally {
        if (rs != null) rs.close()
        stmt.close()
      }
    }
  }

  // ========================================
  // HBase 数据访问
  // ========================================

  /**
   * 获取HBase连接
   * 使用loan pattern，确保连接在使用后自动关闭
   *
   * @param operation HBase操作函数
   * @tparam T 返回值类型
   * @return 操作结果
   */
  def withHBaseConnection[T](operation: Connection => T): Try[T] = {
    var connection: Connection = null
    try {
      connection = ConnectionFactory.createConnection(ConfigManager.getHBaseConfig)
      Success(operation(connection))
    } catch {
      case e: Exception =>
        logger.error("HBase操作失败", e)
        Failure(e)
    } finally {
      if (connection != null) {
        try {
          connection.close()
        } catch {
          case e: Exception =>
            logger.error("HBase连接关闭失败", e)
        }
      }
    }
  }

  /**
   * 批量写入HBase
   *
   * @param tableKey 表配置键名
   * @param columnFamily 列族名
   * @param records 数据记录列表: (rowKey, Seq[(columnQualifier, value)])
   * @return 成功写入的记录数
   */
  def batchInsertToHBase(tableKey: String, columnFamily: String,
                         records: Seq[(String, Seq[(String, Any)])]): Try[Int] = {
    withHBaseConnection { conn =>
      val tableName = ConfigManager.getHBaseTableName(tableKey)
      val table = conn.getTable(TableName.valueOf(tableName))
      try {
        val cfBytes = Bytes.toBytes(columnFamily)
        val puts = records.map { case (rowKey, columns) =>
          val put = new Put(Bytes.toBytes(rowKey))
          columns.foreach { case (qualifier, value) =>
            value match {
              case v: Int => put.addColumn(cfBytes, Bytes.toBytes(qualifier), Bytes.toBytes(v))
              case v: Long => put.addColumn(cfBytes, Bytes.toBytes(qualifier), Bytes.toBytes(v))
              case v: Double => put.addColumn(cfBytes, Bytes.toBytes(qualifier), Bytes.toBytes(v))
              case v: String => put.addColumn(cfBytes, Bytes.toBytes(qualifier), Bytes.toBytes(v))
              case v: Float => put.addColumn(cfBytes, Bytes.toBytes(qualifier), Bytes.toBytes(v))
              case v: Boolean => put.addColumn(cfBytes, Bytes.toBytes(qualifier), Bytes.toBytes(v))
              case v: Array[Byte] => put.addColumn(cfBytes, Bytes.toBytes(qualifier), v)
              case other =>
                put.addColumn(cfBytes, Bytes.toBytes(qualifier), Bytes.toBytes(other.toString))
            }
          }
          put
        }
        table.put(puts.asJava)
        records.size
      } finally {
        table.close()
      }
    }
  }

  // ========================================
  // 数据验证工具
  // ========================================

  /**
   * 验证传感器值是否在合理范围内
   */
  def validateSensorValue(sensorType: String, value: Double): Boolean = {
    val (min, max) = ConfigManager.getSensorRange(sensorType)
    if (value < min || value > max) {
      logger.warn(s"传感器值超出合理范围: type=$sensorType, value=$value, range=[$min, $max]")
      false
    } else {
      true
    }
  }

  /**
   * 验证销售价格是否在合理范围内
   */
  def validateSalePrice(price: Double): Boolean = {
    val (min, max) = ConfigManager.getSalePriceRange
    if (price < min || price > max) {
      logger.warn(s"销售价格超出合理范围: price=$price, range=[$min, $max]")
      false
    } else {
      true
    }
  }

  /**
   * 验证客户年龄是否在合理范围内
   */
  def validateCustomerAge(age: Int): Boolean = {
    val (min, max) = ConfigManager.getCustomerAgeRange
    if (age < min || age > max) {
      logger.warn(s"客户年龄超出合理范围: age=$age, range=[$min, $max]")
      false
    } else {
      true
    }
  }

  /**
   * 通用空值检查
   */
  def requireNonEmpty(value: String, fieldName: String): Unit = {
    require(value != null && value.trim.nonEmpty, s"$fieldName 不能为空")
  }
}

package com.sparklearning.auto.producer

import com.sparklearning.auto.config.ConfigManager
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.slf4j.{Logger, LoggerFactory}

import scala.util.Random

/**
 * 销售数据生产者 - 模拟销售数据发送到Kafka
 *
 * 企业级改进：
 * 1. 配置从ConfigManager获取，不硬编码
 * 2. 使用try-finally确保Producer资源释放
 * 3. 添加数据验证逻辑
 * 4. 添加日志记录
 */
object SalesDataProducer {

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

  def main(args: Array[String]): Unit = {
    val producerProps = ConfigManager.getKafkaProducerConfig
    val topic = ConfigManager.getKafkaTopic("sales-data")
    val producer = new KafkaProducer[String, String](producerProps)
    val random = new Random()

    val brands = Array("比亚迪", "吉利", "长安", "奇瑞", "哈弗", "红旗",
                       "蔚来", "理想", "小鹏", "问界", "极氪", "领克")
    val models = Array("秦PLUS", "星瑞", "UNI-V", "瑞虎8", "H6", "H5",
                        "ES6", "L7", "P7", "M5", "001", "08")
    val regions = Array("北京", "上海", "广州", "深圳", "杭州", "成都", "武汉", "西安",
                        "南京", "重庆", "苏州", "天津")
    val paymentMethods = Array("现金", "贷款", "分期付款", "融资租赁")

    try {
      var sentCount = 0L
      while (true) {
        val saleId = System.currentTimeMillis()
        val vehicleId = 1000 + random.nextInt(9000)
        val customerId = 100 + random.nextInt(900)
        val dealerId = 10 + random.nextInt(90)
        val year = 2023 + random.nextInt(4)
        val month = 1 + random.nextInt(12)
        val day = 1 + random.nextInt(28)
        val saleDate = f"$year-$month%02d-$day%02d"
        val salePrice = 100000.0 + random.nextInt(900000)
        val paymentMethod = paymentMethods(random.nextInt(paymentMethods.length))
        val region = regions(random.nextInt(regions.length))

        // 数据验证
        if (!DataValidator.validateSalePrice(salePrice)) {
          logger.warn(s"跳过无效销售价格: $salePrice")
          Thread.sleep(2000)
          next
        }

        val salesData = s"""{"sale_id":$saleId,"vehicle_id":$vehicleId,"customer_id":$customerId,"dealer_id":$dealerId,"sale_date":"$saleDate","sale_price":$salePrice,"payment_method":"$paymentMethod","region":"$region"}"""

        val record = new ProducerRecord[String, String](topic, null, salesData)
        producer.send(record)

        sentCount += 1
        if (sentCount % 100 == 0) {
          logger.info(s"已发送销售数据: $sentCount 条")
        }
        Thread.sleep(2000)
      }
    } catch {
      case e: InterruptedException =>
        logger.info("销售数据生产者被中断，正在关闭...")
      case e: Exception =>
        logger.error("销售数据生产者异常", e)
    } finally {
      producer.close()
      logger.info("Kafka Producer已关闭")
    }
  }
}

/**
 * 数据验证工具 - 集中管理数据验证逻辑
 */
object DataValidator {

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

  /**
   * 验证销售价格
   */
  def validateSalePrice(price: Double): Boolean = {
    val (min, max) = ConfigManager.getSalePriceRange
    val valid = price >= min && price <= max
    if (!valid) {
      logger.warn(s"销售价格超出合理范围: price=$price, range=[$min, $max]")
    }
    valid
  }

  /**
   * 验证传感器值
   */
  def validateSensorValue(sensorType: String, value: Double): Boolean = {
    val (min, max) = ConfigManager.getSensorRange(sensorType)
    val valid = value >= min && value <= max
    if (!valid) {
      logger.warn(s"传感器值超出合理范围: type=$sensorType, value=$value, range=[$min, $max]")
    }
    valid
  }

  /**
   * 验证非空字符串
   */
  def validateNonEmpty(value: String, fieldName: String): Boolean = {
    val valid = value != null && value.trim.nonEmpty
    if (!valid) {
      logger.warn(s"$fieldName 值为空")
    }
    valid
  }

  /**
   * 验证数值范围
   */
  def validateRange(value: Double, min: Double, max: Double, fieldName: String): Boolean = {
    val valid = value >= min && value <= max
    if (!valid) {
      logger.warn(s"$fieldName 超出范围: value=$value, range=[$min, $max]")
    }
    valid
  }
}

package com.sparklearning.auto.producer

import com.sparklearning.auto.config.ConfigManager
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.slf4j.{Logger, LoggerFactory}

import scala.util.Random

/**
 * 传感器数据生产者 - 模拟车辆传感器数据发送到Kafka
 *
 * 企业级改进：
 * 1. 配置从ConfigManager获取
 * 2. 传感器值范围验证
 * 3. try-finally确保资源释放
 * 4. 完善的异常处理
 */
object SensorDataProducer {

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

  def main(args: Array[String]): Unit = {
    val producerProps = ConfigManager.getKafkaProducerConfig
    val topic = ConfigManager.getKafkaTopic("sensor-data")
    val producer = new KafkaProducer[String, String](producerProps)
    val random = new Random()

    val sensorTypes = Array("发动机温度", "机油压力", "冷却液温度", "轮胎压力", "燃油液位", "电池电压")

    // 各传感器的正常值范围
    val sensorNormalRanges = Map(
      "发动机温度" -> (70.0, 105.0),
      "机油压力" -> (1.5, 5.0),
      "冷却液温度" -> (75.0, 100.0),
      "轮胎压力" -> (2.0, 2.8),
      "燃油液位" -> (5.0, 95.0),
      "电池电压" -> (11.5, 14.5)
    )

    try {
      var sentCount = 0L
      while (true) {
        val sensorId = 1000 + random.nextInt(9000)
        val vehicleId = 1000 + random.nextInt(9000)
        val timestamp = System.currentTimeMillis()
        val sensorType = sensorTypes(random.nextInt(sensorTypes.length))
        val (normalMin, normalMax) = sensorNormalRanges(sensorType)

        // 90%概率生成正常值，10%概率生成异常值
        val value = if (random.nextDouble() < 0.9) {
          normalMin + random.nextDouble() * (normalMax - normalMin)
        } else {
          // 生成异常值：超出正常范围
          val (globalMin, globalMax) = ConfigManager.getSensorRange(sensorType)
          if (random.nextBoolean()) {
            globalMin + random.nextDouble() * (normalMin - globalMin)
          } else {
            normalMax + random.nextDouble() * (globalMax - normalMax)
          }
        }

        val roundedValue = BigDecimal(value).setScale(1, BigDecimal.RoundingMode.HALF_UP).toDouble
        val status = if (roundedValue < normalMin || roundedValue > normalMax) "异常" else "正常"

        val sensorData = s"""{"sensor_id":$sensorId,"vehicle_id":$vehicleId,"timestamp":$timestamp,"sensor_type":"$sensorType","value":$roundedValue,"status":"$status"}"""

        val record = new ProducerRecord[String, String](topic, String.valueOf(vehicleId), sensorData)
        producer.send(record)

        sentCount += 1
        if (sentCount % 100 == 0) {
          logger.info(s"已发送传感器数据: $sentCount 条")
        }
        Thread.sleep(1000)
      }
    } catch {
      case e: InterruptedException =>
        logger.info("传感器数据生产者被中断，正在关闭...")
      case e: Exception =>
        logger.error("传感器数据生产者异常", e)
    } finally {
      producer.close()
      logger.info("Kafka Producer已关闭")
    }
  }
}

package com.sparklearning.auto.storage

import com.sparklearning.auto.config.ConfigManager
import com.sparklearning.auto.dataaccess.DataAccessLayer
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.types._
import org.slf4j.{Logger, LoggerFactory}

/**
 * 传感器数据写入HBase
 *
 * 企业级改进：
 * 1. SparkSession使用try-finally确保释放
 * 2. HBase连接配置从ConfigManager获取
 * 3. 数据验证：过滤无效传感器值
 * 4. 使用foreachBatch + DataAccessLayer写入HBase
 * 5. 添加降级策略：HBase不可用时暂存到HDFS
 */
object SensorDataToHBase {

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

  def main(args: Array[String]): Unit = {
    try {
      val spark = SparkSession.builder()
        .appName("SensorDataToHBase")
        .master(ConfigManager.getSparkMaster)
        .config("spark.sql.shuffle.partitions", ConfigManager.getSparkShufflePartitions)
        .getOrCreate()

      import spark.implicits._

      // 从Kafka读取传感器数据
      val sensorStream = spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", ConfigManager.getKafkaBootstrapServers)
        .option("subscribe", ConfigManager.getKafkaTopic("sensor-data"))
        .option("startingOffsets", "latest")
        .option("maxOffsetsPerTrigger", "1000")
        .load()

      // 定义Schema
      val schema = new StructType()
        .add("sensor_id", IntegerType)
        .add("vehicle_id", IntegerType)
        .add("timestamp", LongType)
        .add("sensor_type", StringType)
        .add("value", DoubleType)
        .add("status", StringType)

      // 解析JSON数据 + 数据验证
      val sensorData = sensorStream
        .selectExpr("CAST(value AS STRING)")
        .select(from_json($"value", schema).as("data"))
        .select("data.*")
        .filter($"data".isNotNull) // 过滤解析失败的数据
        .filter($"sensor_id".isNotNull && $"vehicle_id".isNotNull && $"timestamp".isNotNull)
        .filter($"sensor_type".isNotNull && $"value".isNotNull && $"status".isNotNull)

      // 写入HBase（带降级策略）
      val query = sensorData.writeStream
        .foreachBatch { (batchDF: org.apache.spark.sql.DataFrame, batchId: Long) =>
          logger.info(s"处理批次: $batchId, 记录数: ${batchDF.count()}")

          try {
            // 尝试写入HBase
            writeBatchToHBase(batchDF, batchId)
          } catch {
            case e: Exception =>
              logger.error(s"批次 $batchId 写入HBase失败，启动降级策略", e)
              // 降级：将数据暂存到HDFS
              fallbackToHDFS(batchDF, batchId)
          }
        }
        .outputMode(OutputMode.Update())
        .option("checkpointLocation", ConfigManager.getSparkCheckpointDir + "/sensor-to-hbase")
        .start()

      query.awaitTermination()

    } catch {
      case e: Exception =>
        logger.error("SensorDataToHBase运行异常", e)
        throw e
    } finally {
      val s = SparkSession.getActiveSession.orNull
      if (s != null) {
        s.stop()
        logger.info("SparkSession已关闭")
      }
    }
  }

  /**
   * 将批次数据写入HBase
   */
  private def writeBatchToHBase(batchDF: org.apache.spark.sql.DataFrame, batchId: Long): Unit = {
    val records = batchDF.collect().flatMap { row =>
      val sensorType = row.getAs[String]("sensor_type")
      val value = row.getAs[Double]("value")

      // 数据验证
      if (!DataAccessLayer.validateSensorValue(sensorType, value)) {
        logger.warn(s"批次 $batchId: 跳过无效传感器数据, type=$sensorType, value=$value")
        None
      } else {
        val rowKey = s"${row.getAs[Long]("timestamp")}-${row.getAs[Int]("sensor_id")}"
        val columns = Seq(
          ("vehicle_id", row.getAs[Int]("vehicle_id").asInstanceOf[Any]),
          ("sensor_type", sensorType.asInstanceOf[Any]),
          ("value", value.asInstanceOf[Any]),
          ("status", row.getAs[String]("status").asInstanceOf[Any])
        )
        Some((rowKey, columns))
      }
    }

    if (records.nonEmpty) {
      val result = DataAccessLayer.batchInsertToHBase("sensor-data", "cf", records)
      result match {
        case Success(count) => logger.info(s"批次 $batchId 成功写入HBase: $count 条记录")
        case Failure(e) => throw e
      }
    }
  }

  /**
   * 降级策略：将数据暂存到HDFS
   */
  private def fallbackToHDFS(batchDF: org.apache.spark.sql.DataFrame, batchId: Long): Unit = {
    if (!ConfigManager.isDegradationEnabled) {
      logger.warn("降级策略未启用，数据可能丢失")
      return
    }

    val fallbackPath = s"${ConfigManager.getFallbackDataPath}/sensor_data/batch_$batchId"
    logger.info(s"降级策略：将批次 $batchId 数据暂存到 $fallbackPath")

    try {
      batchDF.write
        .mode("overwrite")
        .json(fallbackPath)
      logger.info(s"批次 $batchId 数据已暂存到HDFS: $fallbackPath")
    } catch {
      case e: Exception =>
        logger.error(s"降级策略执行失败，批次 $batchId 数据可能丢失", e)
    }
  }
}

package com.sparklearning.auto.storage

import com.sparklearning.auto.config.ConfigManager
import com.sparklearning.auto.dataaccess.DataAccessLayer
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.types._
import org.slf4j.{Logger, LoggerFactory}

/**
 * 销售数据写入MySQL
 *
 * 企业级改进：
 * 1. SparkSession使用try-finally确保释放
 * 2. MySQL连接配置从ConfigManager获取，不硬编码密码
 * 3. 数据验证：过滤无效价格和空值
 * 4. 添加降级策略：MySQL不可用时暂存到HDFS
 */
object SalesDataToMySQL {

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

  def main(args: Array[String]): Unit = {
    try {
      val spark = SparkSession.builder()
        .appName("SalesDataToMySQL")
        .master(ConfigManager.getSparkMaster)
        .config("spark.sql.shuffle.partitions", ConfigManager.getSparkShufflePartitions)
        .getOrCreate()

      import spark.implicits._

      // 从Kafka读取销售数据
      val salesStream = spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", ConfigManager.getKafkaBootstrapServers)
        .option("subscribe", ConfigManager.getKafkaTopic("sales-data"))
        .option("startingOffsets", "latest")
        .option("maxOffsetsPerTrigger", "500")
        .load()

      // 定义Schema
      val schema = new StructType()
        .add("sale_id", LongType)
        .add("vehicle_id", IntegerType)
        .add("customer_id", IntegerType)
        .add("dealer_id", IntegerType)
        .add("sale_date", StringType)
        .add("sale_price", DoubleType)
        .add("payment_method", StringType)
        .add("region", StringType)

      // 解析JSON数据 + 数据验证
      val salesData = salesStream
        .selectExpr("CAST(value AS STRING)")
        .select(from_json($"value", schema).as("data"))
        .select("data.*")
        .filter($"data".isNotNull)
        .filter($"sale_id".isNotNull && $"vehicle_id".isNotNull && $"customer_id".isNotNull)
        .filter($"sale_date".isNotNull && $"sale_price".isNotNull)
        .filter($"payment_method".isNotNull && $"region".isNotNull)
        .filter($"sale_price" > 0) // 价格必须为正数
        .withColumn("sale_date", to_date($"sale_date"))

      // 写入MySQL（带降级策略）
      val query = salesData.writeStream
        .foreachBatch { (batchDF: org.apache.spark.sql.DataFrame, batchId: Long) =>
          logger.info(s"处理批次: $batchId, 记录数: ${batchDF.count()}")

          try {
            // 使用ConfigManager获取MySQL配置，不硬编码密码
            val tableName = ConfigManager.getDatabaseTable("sales")
            val mysqlOptions = ConfigManager.getMySQLOptions(tableName)

            batchDF.write
              .format("jdbc")
              .options(mysqlOptions)
              .mode("append")
              .save()

            logger.info(s"批次 $batchId 成功写入MySQL")
          } catch {
            case e: Exception =>
              logger.error(s"批次 $batchId 写入MySQL失败，启动降级策略", e)
              fallbackToHDFS(batchDF, batchId)
          }
        }
        .outputMode(OutputMode.Append())
        .option("checkpointLocation", ConfigManager.getSparkCheckpointDir + "/sales-to-mysql")
        .start()

      query.awaitTermination()

    } catch {
      case e: Exception =>
        logger.error("SalesDataToMySQL运行异常", e)
        throw e
    } finally {
      val s = SparkSession.getActiveSession.orNull
      if (s != null) {
        s.stop()
        logger.info("SparkSession已关闭")
      }
    }
  }

  /**
   * 降级策略：将数据暂存到HDFS
   */
  private def fallbackToHDFS(batchDF: org.apache.spark.sql.DataFrame, batchId: Long): Unit = {
    if (!ConfigManager.isDegradationEnabled) {
      logger.warn("降级策略未启用，数据可能丢失")
      return
    }

    val fallbackPath = s"${ConfigManager.getFallbackDataPath}/sales_data/batch_$batchId"
    logger.info(s"降级策略：将批次 $batchId 数据暂存到 $fallbackPath")

    try {
      batchDF.write
        .mode("overwrite")
        .json(fallbackPath)
      logger.info(s"批次 $batchId 数据已暂存到HDFS: $fallbackPath")
    } catch {
      case e: Exception =>
        logger.error(s"降级策略执行失败，批次 $batchId 数据可能丢失", e)
    }
  }
}