数据管道文档生成

文档：从PostgreSQL到Hive的数据管道设计

1. 目标与范围
- 将PostgreSQL业务数据稳定、可扩展地传输到Hive数据仓库，用于离线分析与报表。
- 支持全量初始化与增量同步（基于时间戳或CDC）。
- 保证数据质量、可追踪性与安全合规。

2. 架构概述
- 批处理方案（适用于每日T+1或小时级同步）
  - 组件：Airflow（编排）、Spark（JDBC抽取与转换）、Hive（存储/Metastore）、HDFS（数据湖）
  - 流程：JDBC读取PostgreSQL → 转换（类型映射、分区列生成）→ 写入Hive（Parquet/ORC）
- CDC流式方案（适用于准实时）
  - 组件：Debezium（WAL变更捕获）→ Kafka（消息缓冲）→ Spark Structured Streaming（增量处理）→ Hive/Hudi/Iceberg（存储、支持Upsert与Schema演进）
  - 流程：PostgreSQL变更→Kafka→流式处理→Hudi表Upsert→Hive查询

3. 数据模型与类型映射
- 推荐存储格式：Parquet（列式、压缩、高效查询）
- 常见类型映射：
  - INTEGER/BIGINT/SMALLINT → Hive INT/BIGINT
  - NUMERIC(p,s) → Hive DECIMAL(p,s)
  - BOOLEAN → Hive BOOLEAN
  - TEXT/VARCHAR → Hive STRING
  - TIMESTAMP/TIMESTAMPTZ → Hive TIMESTAMP（统一转换为UTC）
  - DATE → Hive DATE（若Hive版本不支持DATE，降级为STRING）
  - UUID → STRING
  - JSON/JSONB → STRING（必要时进入ODS层，后续在DWD层进行扁平化）
  - BYTEA → BINARY
- 约束：Hive不强制主键与唯一约束；如需Upsert语义，使用Hudi/Iceberg。

4. 存储与分区设计
- 分层建议：
  - ODS（原始层）：结构与字段尽量贴近源表，轻转换。
  - DWD（明细层）：清洗、类型规范、业务分区。
- 分区列：按事件日期或更新时间（如 event_date 或 updated_at），便于裁剪与重跑。
- 表定义示例（Hive/ODS层，Parquet，外表）：
  - CREATE EXTERNAL TABLE ods.orders (
    order_id BIGINT,
    user_id BIGINT,
    amount DECIMAL(18,2),
    status STRING,
    created_at TIMESTAMP,
    updated_at TIMESTAMP
  )
  PARTITIONED BY (dt STRING)
  STORED AS PARQUET
  LOCATION 'hdfs:///warehouse/ods/orders';

5. 批处理流程（全量与增量）
- 全量初始化
  - 按表并行抽取，限制并发连接数，使用分片列实现并行（主键或自增ID）。
  - 流程：
    1) 读取PostgreSQL最小/最大主键，计算分片范围。
    2) JDBC并行拉取数据，统一到UTC时间，生成分区列 dt。
    3) 先写入临时HDFS路径，完成后原子rename到目标分区路径。
    4) 修复分区元数据（MS分区注册）。
- 增量同步（两种）
  - 基于时间戳：按 updated_at > 上次水位 的条件拉取，数据重复插入采用幂等策略（覆盖当日分区或Hudi Upsert）。
  - 基于序列/主键：按主键范围 + 变更标识字段组合，保证不漏不重。
- 幂等与回滚
  - 写入临时目录（如 …/staging/dt=YYYY-MM-DD），成功后移动到 …/dt=YYYY-MM-DD。
  - 失败保留临时数据供重试；分区级重跑可清理分区后重写。

6. CDC流式方案（准实时）
- Debezium连接器（PostgreSQL）：
  - 读取WAL，输出到Kafka主题，包含操作类型（c/u/d）、前后镜像。
  - 建议使用只读账号并启用TLS；库级白名单。
- Spark Structured Streaming处理：
  - 解析Kafka消息（JSON/Avro），标准化时间字段与类型。
  - 写入Hudi表以支持Upsert与删除标记。
- Hudi写入选项示例（Scala）：
  - df.write.format("hudi")
    .option("hoodie.table.name", "dwd_orders")
    .option("hoodie.datasource.write.recordkey.field", "order_id")
    .option("hoodie.datasource.write.precombine.field", "updated_at")
    .option("hoodie.datasource.write.table.type", "COPY_ON_WRITE")
    .option("hoodie.datasource.write.operation", "upsert")
    .option("hoodie.datasource.hive_sync.enable", "true")
    .option("hoodie.datasource.hive_sync.database", "dwd")
    .option("hoodie.datasource.hive_sync.table", "orders")
    .option("hoodie.datasource.hive_sync.partition_fields", "dt")
    .save("hdfs:///warehouse/dwd/orders")

7. 编排与调度
- 使用Airflow管理依赖与重试：
  - DAG结构：extract_postgres → transform → load_hive → quality_check → success
  - 失败策略：任务级重试、分区级重跑；告警通知。
- 运行窗口：避免与PostgreSQL业务高峰冲突；设置最大连接与查询超时。

8. 代码与命令示例
- Spark JDBC批量抽取（Scala）
  - val jdbcUrl = "jdbc:postgresql://pg-host:5432/appdb"
    val props = new java.util.Properties()
    props.setProperty("user", "readonly_user")
    props.setProperty("password", "******")
    props.setProperty("driver", "org.postgresql.Driver")
    // 先查询主键最小/最大值
    val bounds = spark.read.jdbc(jdbcUrl, "(select min(order_id) as min_id, max(order_id) as max_id from public.orders) as t", props).first
    val df = spark.read.format("jdbc")
      .option("url", jdbcUrl)
      .option("dbtable", "public.orders")
      .option("user", "readonly_user")
      .option("password", "******")
      .option("driver", "org.postgresql.Driver")
      .option("fetchsize", 10000)
      .option("partitionColumn", "order_id")
      .option("lowerBound", bounds.getLong(0))
      .option("upperBound", bounds.getLong(1))
      .option("numPartitions", 16)
      .load()
      .withColumn("dt", to_date(col("updated_at"))) // 分区列
    df.write.mode("overwrite")
      .format("parquet")
      .partitionBy("dt")
      .saveAsTable("ods.orders")
- Sqoop（适用于较简单的全量场景）
  - sqoop import \
    --connect jdbc:postgresql://pg-host:5432/appdb \
    --username readonly_user --password ****** \
    --table public.orders \
    --hive-import --hive-table ods.orders \
    --hive-overwrite \
    --as-parquetfile \
    --split-by order_id \
    --num-mappers 8
  - 说明：Sqoop会将数据以Parquet导入HDFS并创建/覆盖对应Hive表。复杂增量与类型规范建议使用Spark。
- Kafka→Spark Streaming（简化示例）
  - spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", "kafka:9092")
    .option("subscribe", "dbserver1.public.orders")
    .load()
    // 解析、清洗后按Hudi选项写入（参考第6节）

9. 数据质量与校验
- 行数对账：源端计数与目标端分区计数比对（允许一定延迟）。
- 字段级校验：非空、范围、正则；金额精度一致性。
- 采样校验：随机抽样对比关键字段值。
- 完整性：确保所有分区均写入成功；Hive元数据与HDFS文件一致。

10. 安全与合规
- PostgreSQL连接启用TLS；只读账号、最小权限。
- Hive/Kerberos认证；Ranger进行细粒度访问控制。
- 敏感字段脱敏或加密存储；遵守数据生命周期与合规要求。
- 审计与血缘：结合Apache Atlas记录表级/字段级血缘。

11. 运维与监控
- 指标：吞吐量、延迟、失败率、重试次数、分区耗时。
- 日志：Airflow任务日志、Spark应用日志、Kafka消费者延迟。
- 存储：分区大小、文件数、元数据膨胀（适度合并小文件）。
- 告警：阈值触发（如行数差异过大、延迟超标）。

12. 性能与资源优化
- PostgreSQL侧：合理设置work_mem、并发连接上限、索引覆盖抽取条件（如updated_at索引）。
- JDBC：fetchsize优化；并行分区读取；避免长事务阻塞。
- Hive：选择Parquet+Snappy；适度分区粒度（按日/小时）；启用向量化读取。
- 小文件治理：合并文件（Spark repartition/coalesce；Hudi compaction）。
- 计算资源：Executor数量与内存按表数据量分级配置。

13. 失败恢复与重跑
- 采用检查点/水位管理（批处理记录最后成功dt；流式使用Spark checkpoint与Kafka offsets）。
- 分区级重跑：清理目标分区后重写；CDC场景通过回放offset或时间范围重处理。
- 临时目录保留以支持故障分析与断点续传。

14. 模式变更与Schema演进
- ODS层允许新增列（向后兼容）；慎重删除或类型变更。
- CDC场景使用Hudi/Iceberg的Schema演进能力；对Hive同步开启schema自动更新。
- 变更前进行影响评估与下游通知。

15. 版本与环境建议
- PostgreSQL ≥ 12（WAL与逻辑复制成熟）
- Hive ≥ 3.x，Spark ≥ 3.x
- Debezium ≥ 2.x，Kafka ≥ 2.x
- Hudi ≥ 0.11+ 或 Iceberg ≥ 0.13+（需与Spark/Hive版本兼容）

该设计可覆盖从离线批处理到准实时的主要数据同步需求，确保在数据摄取、转换、存储与检索环节的稳定性、可维护性与合规性。

Kafka-to-HDFS Data Pipeline: Technical Overview

Purpose
Move streaming data from Apache Kafka topics into HDFS for durable storage and downstream batch/interactive analytics. The pipeline ensures reliable ingestion, schema-aware transformation, efficient storage, and query-ready layout.

Architecture
- Producers: Publish events to Kafka topics (partitioned and replicated).
- Stream Processor: Consumes Kafka, validates/transforms records, and writes to HDFS.
  - Option A: Spark Structured Streaming (recommended for flexible transformation and strong fault tolerance).
  - Option B: Kafka Connect HDFS Sink (recommended for minimal transformation pipelines).
- Storage: HDFS (HA NameNode), optionally integrated with Hive/Impala/Presto via external tables on the HDFS paths.
- Observability: Metrics and logs from Kafka, stream processor, and HDFS; consumer lag monitoring.

Data Model and Schema
- Message format: Avro or JSON. Prefer Avro/Protobuf with a schema registry to enforce compatibility.
- Required fields:
  - event_id (string or long) for deduplication.
  - event_time (event-time timestamp).
  - topic, partition, offset (from Kafka) for lineage.
- Versioning: Use backward-compatible schema evolution. Validate all records; route invalid records to a dead-letter queue (DLQ) topic or quarantine HDFS path.

HDFS Storage Design
- File format: Parquet (columnar, splittable, efficient compression; e.g., Snappy).
- Directory layout:
  - Base path: hdfs:///data/lake/raw/
  - Partitioning: topic=..., dt=YYYY-MM-DD, hour=HH (event time, not processing time).
  - Example path: hdfs:///data/lake/raw/topicA/dt=2025-09-26/hour=13/part-0000.parquet
- Target file size: 128–256 MB to reduce small files. Control via micro-batch size and repartitioning (Spark) or flush/rotate settings (Kafka Connect).
- Replication factor: 3 (adjust per cluster SLA).
- Hive/Metastore: Create external tables partitioned by dt and hour; enable automatic partition discovery.

Reliability and Consistency
- Delivery semantics:
  - Spark Structured Streaming: Exactly-once file writes with checkpointing under normal operations. Use deterministic transformations; avoid side effects. Optional deduplication using event_id for at-least-once sources.
  - Kafka Connect HDFS Sink: At-least-once delivery. Connector may produce duplicates on failures; mitigate with downstream deduplication keyed by event_id.
- Fault tolerance:
  - Checkpointing: Store offsets and state in HDFS checkpointLocation (Spark).
  - Recovery: Stream processor restarts from last committed offsets or checkpoints.
- Backpressure and rate control:
  - Limit per-trigger offsets (Spark: maxOffsetsPerTrigger).
  - Kafka consumer tuning: fetch sizes, max.poll.interval.ms sized for throughput.

Security
- Kafka: TLS for encryption in transit; SASL (GSSAPI/Kerberos, SCRAM) for authentication; ACLs for authorization.
- HDFS: Kerberos-secured cluster; service principals for stream processor; ACLs on output paths.
- Secrets: Centralized secrets management for credentials; avoid embedding in code/config.

Operations and Monitoring
- Kafka consumer lag: Monitor via group coordinator or tools (e.g., Burrow).
- Stream processor metrics:
  - Spark: Streaming query progress, state/operator metrics, batch durations, input rows per second.
  - Kafka Connect: Task status, sink record rate, failure counts.
- HDFS health: NameNode RPC latency, block reports, DataNode capacity, under-replicated blocks.
- Alerting: Thresholds for lag growth, failed batches/tasks, small-file proliferation, HDFS space.

Retention and Lifecycle
- Kafka topics: Retention sized to accommodate downstream outage windows and replay needs.
- HDFS: Tiered retention; raw zone immutable; curated zone may be compacted or aggregated. Periodic partition compaction to mitigate small files.

Example Implementation A: Spark Structured Streaming (PySpark)
- Characteristics: Flexible transformations, exactly-once file sink with checkpointing, Hive-compatible layout.

from pyspark.sql import SparkSession, functions as F, types as T

spark = (
    SparkSession.builder
    .appName("KafkaToHDFS")
    .getOrCreate()
)

# Define the schema for the Kafka message value (Avro/JSON decoded to JSON here for simplicity)
value_schema = T.StructType([
    T.StructField("event_id", T.StringType(), False),
    T.StructField("event_time", T.TimestampType(), True),
    T.StructField("payload", T.StringType(), True)
])

df = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker1:9092,broker2:9092")
    .option("subscribe", "topicA,topicB")
    .option("startingOffsets", "latest")
    .load()
)

parsed = (
    df.select(
        F.col("topic").cast("string").alias("topic"),
        F.col("timestamp").alias("kafka_ts"),
        F.col("value").cast("string").alias("json")
    )
    .selectExpr("topic", "kafka_ts", "json")
    .withColumn("record", F.from_json(F.col("json"), value_schema))
    .select("topic", "kafka_ts", "record.*")
    .withColumn("event_time", F.coalesce(F.col("event_time"), F.col("kafka_ts")))
    .withColumn("dt", F.to_date(F.col("event_time")))
    .withColumn("hour", F.date_format(F.col("event_time"), "HH"))
    .withWatermark("event_time", "24 hours")  # adjust to lateness tolerance
    # Optional dedup to mitigate at-least-once behavior upstream
    .dropDuplicates(["event_id", "dt", "hour"])
)

query = (
    parsed.writeStream
    .format("parquet")
    .option("path", "hdfs:///data/lake/raw")
    .option("checkpointLocation", "hdfs:///chk/kafka_to_hdfs")
    .partitionBy("topic", "dt", "hour")
    .option("compression", "snappy")
    .option("maxRecordsPerFile", "500000")  # tune to approach target file size
    .outputMode("append")
    .trigger(processingTime="1 minute")
    .start()
)

query.awaitTermination()

Notes:
- For Avro values, decode via spark-avro or use Confluent’s Avro to JSON conversion before parsing.
- Repartition by dt/hour to control file sizes: parsed.repartition(F.col("topic"), F.col("dt"), F.col("hour")).
- Register external Hive tables on hdfs:///data/lake/raw using partition columns.

Example Implementation B: Kafka Connect HDFS Sink (Connector config)
- Characteristics: Simple deployment, low transformation capability, at-least-once delivery.

{
  "name": "hdfs-sink",
  "config": {
    "connector.class": "io.confluent.connect.hdfs.HdfsSinkConnector",
    "tasks.max": "4",
    "topics": "topicA,topicB",
    "hdfs.url": "hdfs://namenode:8020",
    "format.class": "io.confluent.connect.hdfs.parquet.ParquetFormat",
    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "path.format": "topic=${topic}/dt=YYYY-MM-dd/hour=HH",
    "partition.duration.ms": "3600000",
    "flush.size": "100000",
    "rotate.interval.ms": "600000",
    "timezone": "UTC",
    "timestamp.extractor": "RecordField",
    "timestamp.field": "event_time",
    "hadoop.conf.dir": "/etc/hadoop/conf",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "dlq.hdfs"
  }
}

Notes:
- Configure key/value converters (e.g., AvroConverter with Schema Registry) at the Connect worker level or in the connector config if needed.
- The connector writes temporary files and renames on rotation; duplicates can occur on failures. Plan deduplication downstream if required.
- Use Kerberos-enabled Hadoop configs in hadoop.conf.dir for secure clusters.

Testing and Validation
- Functional: Publish test messages; verify HDFS files, partition layout, and schema consistency.
- Fault tolerance: Kill/restart stream processor; validate no missing data and acceptable duplicate behavior.
- Performance: Load test for throughput and end-to-end latency; tune batch duration, partitions, and file sizing.
- Data quality: Validate schema evolution, null handling, and DLQ routing.

Change Management
- Schema changes: Validate in lower environments; enforce compatibility in the registry.
- Topic changes: Update subscriptions and downstream table definitions; maintain backward-compatible paths.

This design provides a robust, secure, and scalable path from Kafka to HDFS with clear trade-offs between Spark and Kafka Connect. Choose the implementation aligned with transformation needs, operational model, and delivery semantics requirements.

MySQL 到 Apache Druid 数据管道设计（简要说明）

目标
- 将MySQL业务表的变更数据持续传输至Druid，用于近实时分析与高并发查询。
- 支持初次全量导入与后续增量（CDC）同步。
- 在Druid中进行时间序列聚合、维度查询和滚动汇总（roll-up），提供稳定的查询性能。

总体架构
- 初次全量：将MySQL数据抽取为列式文件（如Parquet）并通过Druid原生批量摄取（index_parallel）导入历史数据。
- 增量同步：借助Debezium从MySQL Binlog捕获变更，经Kafka传递，Druid通过Kafka Indexing Service进行实时摄取。
- 元数据与控制：使用Airflow或Druid Supervisor管理任务调度与摄取生命周期；通过Druid SQL或原生查询接口提供检索。

数据建模与转换
- 时间列：在业务表中选择或新增事件时间列（event_time）。Druid的__time基于该列，建议统一为UTC。
- 维度与度量：
  - 维度示例：order_id、customer_id、status、region
  - 度量示例：rows（count）、amount_sum（doubleSum），并可扩展为数量、金额、时长等聚合度量。
- Roll-up与粒度：
  - segmentGranularity：DAY（按日分段），适用于多数报表类分析；可按业务需求调整为HOUR/WEEK。
  - queryGranularity：NONE或MINUTE；在实时查询中平衡精度与性能。
  - rollup：true，通过维度+时间粒度聚合提升查询性能与存储效率。
- 主键与幂等：Druid在Kafka摄取为至少一次语义，可能出现重复；通过roll-up或在上游保证幂等（例如使用变更类型与去重逻辑）降低影响。
- 模式演进：Debezium支持Schema变化；使用“unwrap”变换输出扁平结构，Druid通过字段发现或显式维度列表适应新增列。

组件与关键配置
1) Debezium MySQL Connector（Kafka Connect）
- 作用：从MySQL Binlog捕获CDC事件并写入Kafka。配置“unwrap”将事件体扁平化为“after”记录，便于Druid消费。
- 示例（简化）：
{
  "name": "mysql-orders-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "mysql-host",
    "database.port": "3306",
    "database.user": "debezium",
    "database.password": "******",
    "database.server.id": "184054",
    "database.server.name": "mysql",
    "database.include.list": "sales",
    "table.include.list": "sales.orders",
    "include.schema.changes": "false",
    "snapshot.mode": "initial",
    "tombstones.on.delete": "false",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.add.fields": "op,ts_ms",
    "transforms.unwrap.delete.handling.mode": "rewrite",
    "transforms.unwrap.drop.tombstones": "true"
  }
}
- 注意：开启二进制日志（ROW格式），为Debezium用户授予REPLICATION权限；确保表有主键。

2) Druid Kafka 摄取（实时）
- 作用：通过Kafka Indexing Service订阅主题，将JSON事件写入Druid数据源。
- 示例 Supervisor 规范（简化）：
{
  "type": "kafka",
  "dataSchema": {
    "dataSource": "orders",
    "timestampSpec": { "column": "event_time", "format": "auto" },
    "dimensionsSpec": {
      "dimensions": ["order_id","customer_id","status","region"]
    },
    "metricsSpec": [
      { "type": "count", "name": "rows" },
      { "type": "doubleSum", "name": "amount_sum", "fieldName": "amount" }
    ],
    "granularitySpec": {
      "segmentGranularity": "DAY",
      "queryGranularity": "NONE",
      "rollup": true
    }
  },
  "ioConfig": {
    "type": "kafka",
    "topic": "sales.orders",
    "consumerProperties": {
      "bootstrap.servers": "kafka:9092",
      "group.id": "druid-orders-ingest",
      "auto.offset.reset": "earliest"
    },
    "useEarliestOffset": true,
    "inputFormat": { "type": "json" }
  },
  "tuningConfig": {
    "type": "kafka",
    "maxRowsInMemory": 1000000,
    "maxRowsPerSegment": 5000000,
    "intermediatePersistPeriod": "PT10M",
    "maxPendingPersists": 4
  }
}
- 要点：确保event_time为UTC或明确时区解析；如需动态字段发现，可在inputFormat中增加flattenSpec或开启useFieldDiscovery。

3) Druid 批量摄取（初次全量）
- 过程：使用Spark/Flink将MySQL表导出为S3上的Parquet；使用Druid的index_parallel批量导入。
- 示例 Index Task（简化）：
{
  "type": "index_parallel",
  "spec": {
    "dataSchema": {
      "dataSource": "orders",
      "timestampSpec": { "column": "event_time", "format": "auto" },
      "dimensionsSpec": {
        "dimensions": ["order_id","customer_id","status","region"]
      },
      "metricsSpec": [
        { "type": "count", "name": "rows" },
        { "type": "doubleSum", "name": "amount_sum", "fieldName": "amount" }
      ],
      "granularitySpec": {
        "segmentGranularity": "DAY",
        "queryGranularity": "NONE",
        "rollup": true
      }
    },
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": { "type": "s3", "uris": ["s3://bucket/path/orders/*.parquet"] },
      "inputFormat": { "type": "parquet" }
    },
    "tuningConfig": {
      "type": "index_parallel",
      "maxRowsPerSegment": 5000000,
      "maxRowsInMemory": 1000000
    }
  }
}
- 注意：初次全量完成后启动Kafka实时摄取，避免时间范围重叠造成重复；必要时通过时间过滤或去重策略控制。

可靠性与一致性
- Kafka摄取语义：至少一次。任务重启可能重复消费消息；通过roll-up、主键去重或幂等更新策略减轻影响。
- Checkpoint与容错：Druid Kafka Indexing Service通过分段发布与偏移点检查点保障无数据丢失；启用告警监控失败与滞后。
- 回灌与纠错：利用批量摄取对特定时间窗口回灌，随后触发Druid段压缩（compaction）提升查询性能并消除碎片。

性能与容量规划
- 分段与并行：选择合适的segmentGranularity（如DAY）与maxRowsPerSegment，平衡段数量与查询开销。
- Roll-up：通过合理维度组合与queryGranularity，显著降低存储与查询成本。
- 压缩与索引：使用列式格式与位图索引（默认Roaring），定期触发Compaction任务。
- 资源：为MiddleManager/Indexer配置足够的堆与直内存；根据Kafka分区数调节并行摄取任务数。

安全与权限
- MySQL：启用最小权限账户（REPLICATION、SELECT），限制源IP；开启SSL。
- Kafka与Druid：启用SASL/TLS；隔离生产与消费凭据；限制Druid只读访问外部存储。
- Druid集群：保护Coordinator/Overlord/Router端点；基于角色控制提交任务与查询。

运维与监控
- 指标：监控摄取速率、滞后、段发布延时、任务失败率、查询QPS/Latency。
- 日志与告警：为Debezium、Kafka、Druid配置集中式日志；对滞后阈值、失败任务触发告警。
- 自动化：使用Airflow/Argo管理初次全量、回灌与Compaction调度。

验证与测试
- 校验：对比MySQL与Druid在样本时间窗口的行数与聚合结果；检查时间对齐与时区转换。
- 压力测试：模拟峰值写入与查询并发，验证延迟与资源使用。
- 备份与恢复：确认MySQL Binlog保留策略与Druid段备份策略（如对象存储快照）。

检索示例（Druid SQL）
- 示例：按日统计订单金额与数量
SELECT TIME_FLOOR(__time, INTERVAL '1' DAY) AS d, status, SUM(amount_sum) AS total_amount, SUM(rows) AS orders
FROM orders
WHERE __time >= TIMESTAMP '2025-09-01' AND __time < TIMESTAMP '2025-10-01'
GROUP BY d, status
ORDER BY d;

该方案结合批量与实时摄取，满足历史数据回灌与低延迟增量同步的需求，并通过合理的建模、分段与Roll-up提供稳定的查询性能与运维可控性。