使用说明

滑雪场数据摄入管道

概述

该管道从滑雪胜地的Microsoft SQL Server数据库中提取运营数据，并以优化的格式将其加载到云存储（MinIO/S3）中，以便进行分析和报告。

将其视为一个自动化过程：

连接访问度假村的运营数据库
提取物 每日数据（收入、工资、访问、天气等）
变换将其转换为分析就绪格式
负载它变成了一个用于报告和商业智能的数据湖

它收集哪些数据？

管道可以提取以下数据类型（称为“过程”或“procs”）：

数据类型	描述	业务用途
收入	销售和收入指标	跟踪每日销售业绩，识别趋势
工资单	每小时员工时间和工资	监控合同工的劳动力成本
工资单	带薪员工薪酬	跟踪固定人工成本
工资单历史记录	历史工资记录	访问淡季工资数据
预算	预算信息	比较实际预算与计划预算
访问	访客访问指标	了解客流量和出席情况
天气	雪和天气数据	将天气与运营/收入相关联

您可以选择要提取的数据类型（请参见配置在......下面

运作原理

高级流程

┌─────────────────┐
│  SQL Server DB  │  ← Your operational database
│  (MCP Database) │
└────────┬────────┘
         │
         │ 1. Connect & Extract
         │    (Run stored procedures)
         ▼
┌─────────────────┐
│  Data Pipeline  │  ← This script
│  (Python)       │
└────────┬────────┘
         │
         │ 2. Transform & Process
         │    (Clean, aggregate, add metadata)
         ▼
┌─────────────────┐
│  Parquet Files  │  ← Temporary local storage
│  (Local)        │
└────────┬────────┘
         │
         │ 3. Upload & Store
         │    (Copy to cloud)
         ▼
┌─────────────────┐
│  MinIO / S3     │  ← Cloud data lake
│  (Object Store) │
└─────────────────┘

详细步骤

环境设置
- 验证所有必需的配置（数据库凭据、S3详细信息、度假村信息、日期）
- 安装必要的软件依赖项（ODBC驱动程序、Python库）
日期范围计算
- 用途 DATE_START （必填）以及 DATE_END （可选）
- 如果 DATE_END 未提供：
  - 如果当前UTC时间>=15:00，则使用今天的日期
  - 如果当前UTC时间<15:00，则使用昨天的日期
- 这确保了完整的数据可用性（数据截止时间为UTC下午3点）
数据库连接
- 连接到Microsoft SQL Server数据库
- 使用SQL Server的ODBC驱动程序18（行业标准数据库连接器）
每日数据提取 （每天循环）
- 对于范围内的每个日期：
  - 为您的度假胜地运行选定的存储过程（SQL查询）
  - 以行和列的形式检索数据
  - 通过自动重试处理连接问题（3次尝试，30秒延迟）
数据处理
- 收入、预算、访问量、天气：
  - 最少的处理-按原样提取数据
- 对于工资单：
  - 将日期与 ACTIVE_PAYROLL_DATE
  - 主动工资单 （日期>=活动日历日期）：
    - 跑 payroll （小时）+ payroll_salary 存储过程
    - 计算： (hours × rate) + dollar_amount 每个部门
    - 合并为聚合 processed_payroll
  - 历史工资单 （日期<ACTIVE_PAYROLL_date）：
    - 用途 payroll_history 存储过程
    - 提取预先汇总的历史总计
数据格式
- 转换为 拼花格式 （压缩、柱状存储）
- 添加元数据列：
  - _meta_proc：运行了哪个程序
  - _meta_resort：度假村名称
  - _meta_date：YYYY-MM-DD格式的日期
  - _meta_rowcount：记录数
  - fetchedAt：提取数据时（UTC时间戳）
文件组织
- 按以下步骤构建文件→ 度假胜地→ date
- 例子： proc=revenue/resort=purgatory/date_start=2024-11-01/date_end=2024-11-01/part-000.parquet
- 这种结构使分析工具能够进行高效过滤
云端上传
- 将整理好的文件上传到MinIO/S3
- 使用逐日期上传以提高可靠性
清理
- 成功上传后删除本地文件
- 将云存储作为唯一的真相来源

先决条件

所需访问权限

Microsoft SQL Server数据库凭据
具有读/写权限的MinIO或S3存储桶

软件需求

操作系统：Linux（基于Debian）
python：3.x
数据库驱动：SQL Server的ODBC驱动程序18（自动安装）

配置

管道完全通过以下方式配置 环境变量。无需更改代码！

所有环境变量

# ============================================
# REQUIRED VARIABLES
# ============================================

# Database Configuration
MCP_DB_USERNAME              # SQL Server username
MCP_DB_PASSWORD              # SQL Server password
MCP_DB_SERVER                # SQL Server hostname/IP
MCP_DB_NAME                  # Database name

# MinIO/S3 Configuration
MINIO_ENDPOINT               # MinIO endpoint URL (e.g., http://minio.minio.svc:9000)
MINIO_BUCKET                 # S3/MinIO bucket name
AWS_ACCESS_KEY_ID            # MinIO access key
AWS_SECRET_ACCESS_KEY        # MinIO secret key

# Resort Configuration
RESORT_NAME                  # Resort name (e.g., "Big Sky")
RESORT_DB_NAME               # Resort-specific database name
RESORT_GROUP_NUM             # Resort group number (must be integer, e.g., 46)

# Date Configuration
DATE_START                   # Start date in YYYY-MM-DD format
ACTIVE_PAYROLL_DATE          # Cutoff date for active vs historical payroll (YYYY-MM-DD)
                             # Dates >= this use active payroll processing
                             # Dates < this use historical payroll data

# ============================================
# OPTIONAL VARIABLES (with defaults)
# ============================================

# Database Configuration
MCP_DB_PORT=1433             # SQL Server port

# MinIO/S3 Configuration
MINIO_PREFIX=mcp_parquet     # S3 prefix/folder for organized storage
AWS_DEFAULT_REGION=us-east-1 # AWS region (for S3 compatibility)

# Date Configuration
DATE_END                     # End date in YYYY-MM-DD format
                             # If not provided:
                             #   - Uses today if UTC time >= 15:00
                             #   - Uses yesterday if UTC time < 15:00

# Processing Configuration
PROCS=revenue,payroll,payroll_salary,payroll_history,budget,visits,weather
                             # Comma-separated list of procedures to run
                             # Valid: revenue, payroll, payroll_salary,
                             #        payroll_history, budget, visits, weather

OUTDIR=/work/out             # Local output directory

# Error Reporting
ERROR_WEBHOOK_URL=https://n8n-v2.mcp.hyperplane.dev/webhook/data-ingestion-errors
                             # Webhook URL for error notifications

用法

运行管道

# 1. Set required environment variables
export MCP_DB_USERNAME="your_username"
export MCP_DB_PASSWORD="your_password"
export MCP_DB_SERVER="db.example.com"
export MCP_DB_NAME="MCP"

export MINIO_ENDPOINT="http://minio.example.com:9000"
export MINIO_BUCKET="data-lake"
export AWS_ACCESS_KEY_ID="minioadmin"
export AWS_SECRET_ACCESS_KEY="minioadmin"

export RESORT_NAME="PURGATORY"
export RESORT_DB_NAME="Purgatory"
export RESORT_GROUP_NUM="46"

export DATE_START="2024-11-01"
export ACTIVE_PAYROLL_DATE="2024-11-01"  # Payroll cutoff date

# 2. (Optional) Customize what to extract
export PROCS="revenue,visits,weather"

# 3. (Optional) Specify end date (defaults to today/yesterday based on time)
export DATE_END="2024-11-30"

# 4. Run the pipeline
./run.sh

示例：仅提取最近几天的收入

# Set required variables
export MCP_DB_USERNAME="..." MCP_DB_PASSWORD="..." MCP_DB_SERVER="..."
export MCP_DB_NAME="MCP"
export MINIO_ENDPOINT="..." MINIO_BUCKET="..."
export AWS_ACCESS_KEY_ID="..." AWS_SECRET_ACCESS_KEY="..."
export RESORT_NAME="PURGATORY" RESORT_DB_NAME="Purgatory" RESORT_GROUP_NUM="46"

# Configure for revenue only, starting 7 days ago
export DATE_START="2024-11-08"
export ACTIVE_PAYROLL_DATE="2024-11-01"
export PROCS="revenue"
./run.sh

示例：提取特定日期范围的所有数据

# Required config (same as above)...

# Extract all data types for November 2024
export DATE_START="2024-11-01"
export DATE_END="2024-11-30"
export ACTIVE_PAYROLL_DATE="2024-11-01"
./run.sh

输出结构

文件组织

数据存储在针对Dremio、Spark或Athena等分析工具优化的分层结构中：

s3://your-bucket/mcp_parquet/
├── proc=revenue/
│   └── resort=purgatory/
│       └── date=2024-11-01/
│           └── data.parquet
├── proc=payroll/
│   └── resort=purgatory/
│       └── date=2024-11-01/
│           └── data.parquet
├── proc=processed_payroll/  ← Combined hourly + salary totals
│   └── resort=purgatory/
│       └── date=2024-11-01/
│           └── data.parquet
...

拼花地板格式的优点

镶木地板 是一种列式存储格式，它：

压缩数据 高效（文件大小更小）
加快查询速度 （仅读取所需列）
善于使用分析工具 （Dremio、Spark、Tableau、Power BI等）

元数据列

每个文件都包含以下用于跟踪的附加列：

_meta_proc          # Which procedure: "revenue", "payroll", etc.
_meta_resort        # Resort name in kebab-case: "lee-canyon", "purgatory"
_meta_date          # Date: "2024-11-01"
_meta_rowcount      # Number of records: "145"
fetchedAt           # Extraction timestamp: "2024-11-01T15:30:00Z"

技术细节

建筑

语言：Python 3.x
数据库：Microsoft SQL Server（通过 pyodbc +ODBC驱动程序18）
数据处理: pandas 对于DataFrames
存储格式：阿帕奇拼花地板（通过 pyarrow)
云存储：通过以下方式兼容S3（MinIO/AWS S3） boto3

关键组件

文件	目的
`run.sh`	编排整个过程的Shell脚本
`main.py`	具有核心数据提取逻辑的Python脚本

数据处理逻辑

工资处理

管道使用 ACTIVE_PAYROLL_DATE 确定处理方法：

主动工资单 （日期>=活动日历日期）：
- 跑 payroll （小时）+ payroll_salary 存储过程
- 计算工资： (hours × rate) + dollar_amount
- 按部门代码汇总总计
- 将两者结合起来 processed_payroll 输出
历史工资单 （日期<ACTIVE_PAYROLL_date）：
- 跑 payroll_history 存储过程
- 直接使用预先汇总的历史总计
- 更快地处理旧数据

日期处理

处理数据 日复一日 可靠性
DATE_START 是必需的， DATE_END 是可选的
如果 DATE_END 未提供：
- 如果UTC时间>=15:00，则使用今天的日期
- 如果UTC时间<15:00，则使用昨天的日期
下午3点的截止时间确保了每日数据的完整可用性
存储过程调用：包括完整的日期时间范围（每天00:00:00到23:59:59）
元数据和文件夹：使用简单的日期格式（YYYY-MM-DD）

错误处理

数据库检索：自动重试失败的数据库调用（3次尝试，延迟30秒）
上传检索结果：S3/MinIO上传重试3次，失败时延迟10秒
上传验证：在删除本地副本之前，验证上传后S3中是否存在每个文件
安全失效：上传失败时保留本地文件，防止数据丢失
错误收集：通过webhook通知跟踪和报告所有错误
验证：启动前检查所有必需的环境变量
日志记录：输出JSON格式的日志以供监控

监控和日志

管道输出 JSON格式的日志 为了便于解析：

{"seasons_to_process": 1, "seasons": [{"start": "2024-11-01", "end": "2025-03-31"}]}
{"processing_date": "2024-11-01", "season": 1, "date_count": 1}
{"writing": "/work/out/proc=revenue/resort=purgatory/.../part-000.parquet", "rows": 245}
{"uploading": "mcp_parquet/proc=revenue/resort=purgatory/.../part-000.parquet"}
{"upload_verified": "mcp_parquet/proc=revenue/resort=purgatory/.../part-000.parquet"}
{"status": "completed", "total_dates_processed": 150, "location": "s3://bucket/mcp_parquet/"}

Webhook错误报告

当发生故障时，管道会自动向n8n webhook发送错误通知。

Webhook URL: https://n8n-v2.mcp.hyperplane.dev/webhook/data-ingestion-errors （可通过以下方式配置 ERROR_WEBHOOK_URL)

发送通知时：

✅ 上传失败（文件无法上传到S3）
✅ 工资处理错误
✅ 任何关键的管道错误

webhook负载示例：

{
  "pipeline": "data_ingestion",
  "resort": "PURGATORY",
  "timestamp": "2024-11-15T14:30:00Z",
  "total_errors": 2,
  "summary": {
    "status": "completed_with_failures",
    "total_dates_processed": 150,
    "successful_dates": 148,
    "failed_dates": ["2024-11-15", "2024-12-03"],
    "failed_count": 2,
    "location": "s3://bucket/mcp_parquet/"
  },
  "errors": [
    {
      "type": "upload_failure",
      "date": "2024-11-15",
      "resort": "PURGATORY",
      "error_message": "Connection timeout to S3",
      "local_files_preserved": "/work/out",
      "timestamp": "2024-11-15T14:28:33Z"
    },
    {
      "type": "payroll_processing_failure",
      "date": "2024-12-03",
      "resort": "PURGATORY",
      "error_message": "Division by zero in department 201",
      "timestamp": "2024-12-03T14:29:45Z"
    }
  ]
}

错误类型：

upload_failure -S3/MinIO上传失败（保留本地文件以供重试）
payroll_processing_failure -工资汇总失败（其他过程继续）

重要提示：

Webhook通知失败不会导致管道崩溃
每个错误都包括时间戳、日期、度假村和详细的错误消息
摘要概述了管道状态和计数

故障排除

常见问题

“缺少[变量名称]”

原因：未设置所需的环境变量
解决方案：在运行之前导出缺少的变量 ./run.sh

“未知进程'xyz'”

原因：中的过程名称无效 PROCS 变量
解决方案：仅使用有效名称： revenue, payroll, payroll_salary, payroll_history, budget, visits, weather

数据库连接超时

原因：无法访问SQL Server
解决方案：验证 MCP_DB_SERVER 和 MCP_DB_PORT，检查网络连接

S3上传失败

原因：MinIO/S3凭据或终结点无效
解决方案：验证 MINIO_ENDPOINT, AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY

“RESORT_GROUP_NUM必须是有效整数”

原因：组号不是数字
解决方案：设置 RESORT_GROUP_NUM 转换为数值（例如。， 46, -1)

常见问题

Q：管道需要多长时间才能运行？ A：取决于日期范围和数据量。一个完整的季节（约150天）通常需要几个小时。

Q：我可以同时经营多个度假村吗？ A：不是一次跑步。为每个度假村分别运行管道（可以使用单独的进程并行化）。

Q：如果管道中途发生故障怎么办？ A：如果S3上传失败，管道会保留本地文件，防止数据丢失。检查错误webhook通知，了解失败日期的详细信息。您可以使用以下命令在特定日期重新运行 DATE_START 和 DATE_END。如果重新运行，可以安全地覆盖成功上传的日期。

Q：我可以运行这个历史数据吗？ A：是的！集 DATE_START 和 DATE_END 任何过去的日期范围。

Q：我需要每天跑步吗？ A：推荐。通常安排为每日作业（例如，通过cron、Airflow或Kubernetes CronJob）以保持数据新鲜。

Q：加载数据后，我可以在哪里查询数据？ A：使用任何支持Parquet/S3的工具：Dremio、AWS Athena、Spark、Presto，甚至pandas/Python。

Q：我如何知道在管道过程中是否发生了错误？ A：错误会实时发送到配置的webhook（n8n）。检查n8n工作流中的错误通知，包括错误类型、日期和消息等详细信息。

Q：如果webhook关闭了怎么办？ A：管道继续正常完成。Webhook通知失败会被记录下来，但不会影响数据处理或管道状态。

Q：如果S3上传失败，我的本地文件是否安全？ A：是的！只有在成功上传验证后，才会删除本地文件。如果上传失败，文件将保留在 /work/out （或你的 OUTDIR)用于手动恢复或重试。

Q：什么是ACTIVE_PAYROLL_DATE，如何设置它？ A. ACTIVE_PAYROLL_DATE 是决定工资处理方法的截止日期。对于日期>=此值，管道使用详细的活动工资单存储过程（小时+工资）。对于小于此值的日期，它使用历史工资单数据。通常设置为当前工资期的开始（例如，当前滑雪季节的开始，如 2024-11-01).

Q：如果未提供DATE_END，会发生什么？ A：管道会根据当前UTC时间自动确定结束日期。如果是15:00 UTC或更晚，则使用今天的日期。如果在15:00 UTC之前，则使用昨天的日期。这可确保您处理完整的日常数据。

使用说明

滑雪场数据摄入管道

概述

它收集哪些数据？

运作原理

高级流程

详细步骤

先决条件

所需访问权限

软件需求

配置

所有环境变量

用法

运行管道

示例：仅提取最近几天的收入

示例：提取特定日期范围的所有数据

输出结构

文件组织

拼花地板格式的优点

元数据列

技术细节

建筑

关键组件

数据处理逻辑

工资处理

日期处理

错误处理

监控和日志

Webhook错误报告

故障排除

常见问题

常见问题

相关 MCP

地理数据转换服务

AI数据库连接协议

Excel转JSON工具

CSV文件管理服务器

数据管道管理服务器

Arpeio数据工具MCP服务器