数字档案馆系统数据分析实战：从数据采集到可视化全流程解析

一、数据采集与预处理

1.1 确定数据源与采集方式

数字档案馆系统的核心数据通常存储在以下位置：

• 关系型数据库：MySQL/PostgreSQL中的档案元数据表
• 文件存储系统：档案原文的PDF、图像文件
• 日志文件：用户访问日志、系统操作日志
• 外部系统接口：与其他业务系统的数据交换接口

使用Python进行数据采集的完整示例：

import pandas as pd
import pymysql
from sqlalchemy import create_engine
数据库连接配置
db_config = {
'host': 'localhost',
'port': 3306,
'user': 'archive_user',
'password': 'your_password',
'database': 'digital_archive'
}
创建数据库连接
engine = create_engine(
f"mysql+pymysql://{db_config['user']}:{db_config['password']}"
f"@{db_config['host']}:{db_config['port']}/{db_config['database']}"
)
查询档案元数据
query = """
SELECT
archive_id,
title,
create_date,
category,
storage_location,
access_count,
last_access_time
FROM archive_metadata
WHERE status = 'active'
"""
df_metadata = pd.read_sql(query, engine)

1.2 数据清洗与标准化

档案数据清洗的关键步骤：

• 处理缺失值：对关键字段进行填充或标记
• 日期格式统一：将不同格式的日期转换为标准格式
• 分类标准化：统一分类编码和名称

数据清洗代码实现：

 处理缺失值
df_metadata['category'] = df_metadata['category'].fillna('未分类')
日期格式标准化
df_metadata['create_date'] = pd.to_datetime(
df_metadata['create_date'],
errors='coerce'
)
分类编码标准化
category_mapping = {
'文书': 'WS',
'科技': 'KJ',
'基建': 'JJ',
'会计': 'KJ'
}
df_metadata['category_code'] = df_metadata['category'].map(category_mapping)

二、数据分析指标体系构建

2.1 核心业务指标

构建数字档案馆的四大分析维度：

• 档案利用分析：访问频次、检索成功率
• 档案管理分析：入库效率、完整率
• 用户行为分析：用户活跃度、使用偏好
• 系统性能分析：响应时间、存储使用率

2.2 指标计算实现

 计算月度档案利用率
df_metadata['access_month'] = df_metadata['last_access_time'].dt.to_period('M')
monthly_access = df_metadata.groupby('access_month')['access_count'].sum()
计算分类访问热度
category_access = df_metadata.groupby('category')['access_count'].agg(['sum', 'mean', 'count'])
计算档案年龄分布
current_date = pd.Timestamp.now()
df_metadata['archive_age'] = (current_date - df_metadata['create_date']).dt.days
age_distribution = pd.cut(
df_metadata['archive_age'],
bins=[0, 365, 3655, 36510, float('inf')],
labels=['1年内', '1-5年', '5-10年', '10年以上']
).value_counts()

三、数据存储与处理架构

3.1 数据仓库设计

采用星型模型设计数据仓库：

-- 事实表：档案访问事实
CREATE TABLE fact_archive_access (
access_id BIGINT PRIMARY KEY,
archive_id INT,
user_id INT,
access_time TIMESTAMP,
access_type VARCHAR(20),
duration_seconds INT,
success_flag BOOLEAN,
FOREIGN KEY (archive_id) REFERENCES dim_archive(archive_id),
FOREIGN KEY (user_id) REFERENCES dim_user(user_id)
);
-- 维度表：档案维度
CREATE TABLE dim_archive (
archive_id INT PRIMARY KEY,
title VARCHAR(500),
category_id INT,
create_date DATE,
security_level VARCHAR(10),
storage_location VARCHAR(200)
);

3.2 ETL流程配置

使用Apache Airflow配置每日ETL任务：

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
default_args = {
'owner': 'archive_analytics',
'depends_on_past': False,
'start_date': datetime(2024, 1, 1),
'retries': 3,
'retry_delay': timedelta(minutes=5)
}
dag = DAG(
'archive_data_etl',
default_args=default_args,
schedule_interval='0 2   ',   每天凌晨2点执行
catchup=False
)
def extract_data(context):
数据提取逻辑
pass
def transform_data(context):
数据转换逻辑
pass
def load_data(context):
数据加载逻辑
pass
extract_task = PythonOperator(
task_id='extract_archive_data',
python_callable=extract_data,
dag=dag
)
transform_task = PythonOperator(
task_id='transform_archive_data',
python_callable=transform_data,
dag=dag
)
load_task = PythonOperator(
task_id='load_to_data_warehouse',
python_callable=load_data,
dag=dag
)
extract_task >> transform_task >> load_task

四、数据分析与可视化

4.1 使用Pandas进行深度分析

 档案访问时间模式分析
df_access['access_hour'] = df_access['access_time'].dt.hour
hourly_pattern = df_access.groupby('access_hour').size()
用户访问行为聚类
from sklearn.cluster import KMeans
user_features = df_access.pivot_table(
index='user_id',
columns='access_type',
values='access_count',
aggfunc='sum',
fill_value=0
)
kmeans = KMeans(n_clusters=4, random_state=42)
user_features['cluster'] = kmeans.fit_predict(user_features)
档案关联性分析
from mlxtend.frequent_patterns import apriori
archive_matrix = pd.crosstab(
df_access['user_id'],
df_access['archive_id']
).applymap(lambda x: 1 if x > 0 else 0)
frequent_itemsets = apriori(archive_matrix, min_support=0.01, use_colnames=True)

4.2 使用Plotly创建交互式报表

import plotly.express as px
import plotly.graph_objects as go
from plotly.subplots import make_subplots
创建档案访问趋势图
fig1 = px.line(
monthly_access.reset_index(),
x='access_month',
y='access_count',
title='月度档案访问趋势',
labels={'access_count': '访问次数', 'access_month': '月份'}
)
创建档案分类分布旭日图
fig2 = px.sunburst(
df_metadata,
path=['category', 'security_level'],
values='access_count',
title='档案分类与密级分布'
)
创建用户访问热力图
fig3 = go.Figure(data=go.Heatmap(
z=user_features.values,
x=user_features.columns,
y=user_features.index,
colorscale='Viridis'
))
组合图表
fig = make_subplots(
rows=2, cols=2,
subplot_titles=('访问趋势', '分类分布', '用户行为热图', '关联分析'),
specs=[[{'type': 'scatter'}, {'type': 'sunburst'}],
[{'type': 'heatmap'}, {'type': 'table'}]]
)
将图表添加到对应位置
fig.add_trace(fig1.data[0], row=1, col=1)
fig.add_trace(fig2.data[0], row=1, col=2)
fig.add_trace(fig3.data[0], row=2, col=1)
fig.update_layout(height=800, showlegend=False)
fig.show()

五、自动化报表与监控

5.1 报表自动化生成

使用Jupyter Notebook + Papermill实现报表自动化：

 创建报表模板 notebook
archive_daily_report.ipynb 包含完整的数据分析和可视化代码
使用Papermill执行并生成报告
import papermill as pm
pm.execute_notebook(
'archive_daily_report.ipynb',
'archive_report_20240115.ipynb',
parameters={
'report_date': '2024-01-15',
'data_source': 'archive_database',
'output_format': 'html'
}
)
转换为HTML格式
from nbconvert import HTMLExporter
import nbformat
with open('archive_report_20240115.ipynb') as f:
nb = nbformat.read(f, as_version=4)
html_exporter = HTMLExporter()
(body, resources) = html_exporter.from_notebook_node(nb)
with open('archive_report_20240115.html', 'w', encoding='utf-8') as f:
f.write(body)

5.2 关键指标监控告警

 监控配置类
class ArchiveMonitor:
def __init__(self, threshold_config):
self.thresholds = threshold_config
def check_access_decline(self, current_data, historical_data):
"""检查访问量异常下降"""
current_avg = current_data['access_count'].mean()
historical_avg = historical_data['access_count'].mean()
decline_ratio = (historical_avg - current_avg) / historical_avg
if decline_ratio > self.thresholds['access_decline']:
self.send_alert(
f"档案访问量异常下降: {decline_ratio:.1%}",
level='warning'
)
def check_storage_usage(self, storage_info):
"""检查存储空间使用情况"""
usage_ratio = storage_info['used'] / storage_info['total']
if usage_ratio > self.thresholds['storage_warning']:
self.send_alert(
f"存储空间告警: 使用率{usage_ratio:.1%}",
level='critical'
)
def send_alert(self, message, level='info'):
"""发送告警通知"""
alert_data = {
'timestamp': datetime.now().isoformat(),
'level': level,
'message': message,
'system': 'digital_archive'
}
发送到监控系统
requests.post(
'http://monitor.example.com/alerts',
json=alert_data,
headers={'Content-Type': 'application/json'}
)
监控配置
monitor_config = {
'access_decline': 0.3,   访问量下降30%触发告警
'storage_warning': 0.85,   存储使用85%触发告警
'response_time_threshold': 5.0   响应时间超过5秒告警
}
monitor = ArchiveMonitor(monitor_config)

六、部署与维护

6.1 Docker容器化部署

 Dockerfile
FROM python:3.9-slim
WORKDIR /app
安装系统依赖
RUN apt-get update && apt-get install -y \
gcc \
default-libmysqlclient-dev \
&& rm -rf /var/lib/apt/lists/
复制依赖文件
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
复制应用代码
COPY . .
创建数据目录
RUN mkdir -p /data/reports
设置环境变量
ENV PYTHONPATH=/app
ENV DATA_PATH=/data
启动命令
CMD ["python", "scheduler/main.py"]
docker-compose.yml
version: '3.8'
services:
analytics-app:
build: .
ports:
- "8000:8000"
volumes:
- ./config:/app/config
- ./data:/data
environment:
- DB_HOST=archive-db
- DB_PORT=3306
- REDIS_HOST=redis
depends_on:
- archive-db
- redis
archive-db:
image: mysql:8.0
environment:
MYSQL_ROOT_PASSWORD: ${DB_ROOT_PASSWORD}
MYSQL_DATABASE: archive_analytics
volumes:
- db-data:/var/lib/mysql
redis:
image: redis:alpine
volumes:
db-data:

6.2 定期维护任务

 maintenance.py
import schedule
import time
from datetime import datetime, timedelta
def cleanup_old_reports():
"""清理30天前的报表文件"""
import os
import glob
cutoff_date = datetime.now() - timedelta(days=30)
report_files = glob.glob('/data/reports/.html')
for file_path in report_files:
file_time = datetime.fromtimestamp(os.path.getmtime(file_path))
if file_time < cutoff_date:
os.remove(file_path)
print(f"已删除旧报表: {file_path}")
def rebuild_indexes():
"""重建数据库索引"""
from database import get_connection
conn = get_connection()
cursor = conn.cursor()
indexes = [
'fact_archive_access_access_time_idx',
'dim_archive_category_idx',
'dim_user_department_idx'
]
for index in indexes:
cursor.execute(f"REINDEX INDEX CONCURRENTLY {index}")
conn.commit()
cursor.close()
conn.close()
def update_statistics():
"""更新统计信息"""
from database import get_connection
conn = get_connection()
cursor = conn.cursor()
tables = [
'fact_archive_access',
'dim_archive',
'dim_user'
]
for table in tables:
cursor.execute(f"ANALYZE {table}")
conn.commit()
cursor.close()
conn.close()
设置定时任务
schedule.every().day.at("02:00").do(cleanup_old_reports)
schedule.every().sunday.at("03:00").do(rebuild_indexes)
schedule.every().monday.at("01:00").do(update_statistics)
if __name__ == "__main__":
while True:
schedule.run_pending()
time.sleep(60)