# spark-editlog-processor

**Repository Path**: wangjunjios/spark-editlog-processor

## Basic Information

- **Project Name**: spark-editlog-processor
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2025-12-18
- **Last Updated**: 2025-12-18

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Spark SFTP EditLog 解析处理系统

## 项目概述

基于Spark的分布式HDFS EditLog日志解析系统，实现从远程SFTP服务器自动采集EditLog文件，解析关键操作信息并持久化到MySQL数据库。

### 核心功能
- ✅ 自动化监控HDFS元数据操作记录
- ✅ 提供可查询的历史操作审计能力
- ✅ 支持按时间维度快速检索操作记录
- ✅ 增量拉取EditLog文件，按TXID顺序处理
- ✅ 自动时间戳补齐策略
- ✅ MySQL分区表存储，支持海量数据

## 技术栈

| 组件 | 版本 |
|------|------|
| Java | 1.8 |
| Spark | 2.4.8 |
| Hadoop | 2.7.2 |
| MySQL | 5.7+ (支持分区表) |
| Maven | 3.6+ |

## 项目结构

```
SparkEditlogAnysisyTask/
├── pom.xml                                 # Maven配置
├── README.md                               # 项目文档
├── sql/
│   └── create_tables.sql                  # MySQL建表脚本
├── src/main/
│   ├── java/com/spark/editlog/
│   │   ├── model/                         # 数据模型
│   │   │   ├── EditLogFile.java          # EditLog文件模型
│   │   │   ├── EditLogOperation.java     # 操作记录模型
│   │   │   └── ProcessLog.java           # 处理日志模型
│   │   ├── sftp/                          # SFTP管理模块（待实现）
│   │   ├── parser/                        # EditLog解析模块（待实现）
│   │   ├── database/                      # 数据库模块（待实现）
│   │   ├── scheduler/                     # 调度模块（待实现）
│   │   └── EditLogProcessor.java          # 主程序入口（待实现）
│   └── resources/
│       ├── application.properties         # 应用配置
│       └── log4j.properties              # 日志配置
└── .qoder/quests/
    └── spark-sftp-editlog-processing.md  # 详细设计文档
```

## 快速开始

### 1. 环境准备

**前提条件**：
- Java 8
- Maven 3.6+
- Spark 2.4.8运行环境
- MySQL 5.7+ (支持分区表)
- 网络连通：SFTP服务器和MySQL服务器

### 2. 数据库初始化

```bash
# 执行建表脚本
mysql -u root -p < sql/create_tables.sql
```

### 3. 配置修改

编辑 `src/main/resources/application.properties`：

```properties
# SFTP配置
sftp.host=your_sftp_host
sftp.port=22
sftp.username=your_username
sftp.password=your_password
sftp.remote.directory=/hadoop/dfs/name/current

# MySQL配置
mysql.url=jdbc:mysql://your_mysql_host:3306/editlog_db
mysql.username=your_db_username
mysql.password=your_db_password
```

### 4. 编译打包

```bash
mvn clean package
```

生成的JAR包位于：`target/spark-editlog-processor-1.0.0.jar`

### 5. 运行

```bash
# 本地模式
spark-submit \
  --class com.spark.editlog.EditLogProcessor \
  --master local[*] \
  --driver-memory 1g \
  --executor-memory 2g \
  target/spark-editlog-processor-1.0.0.jar

# 集群模式
spark-submit \
  --class com.spark.editlog.EditLogProcessor \
  --master yarn \
  --deploy-mode cluster \
  --driver-memory 1g \
  --executor-memory 2g \
  --num-executors 3 \
  target/spark-editlog-processor-1.0.0.jar
```

## 核心特性

### 1. 增量检测机制
- 自动对比远程SFTP目录和数据库已处理文件列表
- 只下载和处理新增的EditLog文件
- 避免重复处理，节省带宽和计算资源

### 2. TXID顺序处理
- 解析EditLog文件名中的起始TXID
- **严格按TXID升序处理文件**
- 确保HDFS事务的时间顺序性

### 3. 时间戳补齐策略
- **优先级**：TIMESTAMP → MTIME → ATIME
- **补齐逻辑**：使用文件内首个有效TIMESTAMP回填
- **兜底策略**：使用文件下载时间（可配置）

### 4. MySQL分区表
- **分区粒度**：按小时（yyyyMMddHH）
- **分区键**：operation_time
- **动态创建**：写入前自动检测并创建分区
- **查询优化**：利用分区剪枝加速查询

### 5. 异常处理
- SFTP连接失败：3次重试，间隔递增
- 解析失败文件：移动到错误目录保留
- 数据库异常：事务回滚，保留本地文件

## 配置说明

### 关键配置项

| 配置项 | 说明 | 默认值 |
|--------|------|--------|
| scheduler.scan.interval | 扫描间隔（毫秒） | 300000 (5分钟) |
| scheduler.max.concurrent.files | 最大并发文件数 | 1 (顺序处理) |
| timestamp.null.handling | 时间戳缺失处理策略 | DOWNLOAD_TIME |
| mysql.batch.size | 批量写入大小 | 1000 |
| error.retry.max.attempts | 最大重试次数 | 3 |

### 时间戳处理策略

- `DOWNLOAD_TIME`：使用文件下载时间（推荐）
- `FILE_MTIME`：使用EditLog文件修改时间
- `NULL`：保持空值，后续人工处理

## 监控指标

| 指标 | 说明 | 查询方式 |
|------|------|----------|
| 文件处理成功率 | 成功/总数 | 查询process_log表 |
| 平均处理耗时 | 单文件处理时长 | process_end_time - process_start_time |
| 时间戳补齐统计 | backfill_count | 按文件统计 |
| 分区数量 | 当前总分区数 | information_schema.PARTITIONS |

### 监控查询示例

```sql
-- 查询处理统计
SELECT 
    process_status,
    COUNT(*) as file_count,
    SUM(record_count) as total_records,
    SUM(backfill_count) as total_backfills
FROM editlog_process_log
GROUP BY process_status;

-- 查询最近处理的文件
SELECT 
    file_name,
    process_status,
    record_count,
    TIMESTAMPDIFF(SECOND, process_start_time, process_end_time) as duration_seconds
FROM editlog_process_log
ORDER BY process_start_time DESC
LIMIT 10;
```

## 开发进度

### 已完成
- ✅ Maven项目结构和pom.xml
- ✅ 配置文件（application.properties、log4j）
- ✅ 数据模型类（EditLogFile、EditLogOperation、ProcessLog）
- ✅ MySQL建表SQL脚本

### 待实现（核心业务模块）
- ⏳ SFTP管理模块（连接、扫描、增量检测、文件下载）
- ⏳ EditLog解析模块（XML转换、Spark SQL解析）
- ⏳ 数据库模块（连接管理、分区管理、批量写入）
- ⏳ 调度模块（定时任务、流程控制）
- ⏳ 主程序入口和完整流程集成

## 详细设计

完整的设计文档请参考：[设计文档]

设计文档包含：
- 系统架构设计
- 核心模块职责
- 数据流设计
- 异常处理策略
- 性能优化要点
- 部署与运维指南

## 注意事项

1. **TXID顺序至关重要**：必须按TXID升序处理EditLog文件，确保事务顺序性
2. **分区管理**：定期检查分区数量，避免超过MySQL分区上限
3. **磁盘空间**：确保本地缓存目录有足够空间存储临时文件
4. **网络稳定性**：SFTP下载大文件时需要稳定的网络连接
5. **MySQL权限**：确保数据库账号有CREATE、ALTER权限（用于动态创建分区）

## 许可证

内部项目，仅供学习和研究使用。

---

**联系方式**：如有问题请参考设计文档或联系项目维护人员