# Event Stream Ingestion

> 用于新闻/公告/工单流的事件流接入原子 skill，适用于通用行业数据接入场景。

- Skill: `aifinlab/event-stream-ingestion` (Agent Skill)
- Install (CLI): `npx skillmds@latest add aifinlab/event-stream-ingestion`
- Raw SKILL.md: https://api.skillmd.com/api/skills/aifinlab/event-stream-ingestion/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: Coding & Dev Tools
- Author: aifinlab (https://skillmd.com/u/aifinlab)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/aifinlab/event-stream-ingestion

---


# 事件流接入 Skill

## 数据来源

本 Skill 支持多种事件流数据输入格式，核心数据来源包括：

### 1. 事件流类型
- **新闻流**：新闻事件流、资讯事件流
- **公告流**：公告事件流、通知事件流
- **工单流**：工单事件流、任务事件流
- **其他事件流**：日志事件流、监控事件流

### 2. 数据源类型
- **消息队列**：Kafka、RabbitMQ、RocketMQ等
- **API接口**：事件API、Webhook API
- **数据库**：事件数据库、日志数据库
- **文件流**：日志文件、事件文件

### 3. 数据格式要求
- **连接信息**：消息队列连接信息、API端点地址
- **认证信息**：用户名密码、API密钥、Token
- **订阅配置**：事件订阅配置、过滤规则
- **数据格式**：事件数据格式和编码方式

### 4. 事件特征
- **事件频率**：高频事件、低频事件
- **事件规模**：小规模事件、大规模事件
- **事件类型**：单一类型事件、混合类型事件
- **事件实时性**：实时事件、延迟事件

> 说明：本 Skill 不包含事件源配置功能，需要用户提供事件流连接信息。建议事件流稳定可靠，以便进行准确的事件接入。

---

## 功能

本 Skill 提供全面的事件流接入能力，涵盖多种接入功能：

### 1. 事件流连接
- **连接管理**：管理事件流连接
- **连接验证**：验证事件流连接
- **连接监控**：监控事件流状态
- **连接优化**：优化连接性能

### 2. 事件订阅
- **主题订阅**：订阅事件主题
- **过滤订阅**：基于条件过滤订阅
- **多主题订阅**：订阅多个事件主题
- **动态订阅**：动态调整订阅配置

### 3. 事件接收
- **实时接收**：实时接收事件数据
- **批量接收**：批量接收事件数据
- **增量接收**：增量接收事件数据
- **断点续传**：支持断点续传功能

### 4. 事件解析
- **事件解析**：解析事件数据格式
- **事件分类**：对事件进行分类
- **事件标签**：为事件添加标签
- **事件提取**：提取事件关键信息

### 5. 事件处理
- **事件过滤**：过滤不需要的事件
- **事件转换**：转换事件格式
- **事件聚合**：聚合相关事件
- **事件去重**：去除重复事件

### 6. 高级处理功能
- **事件存储**：存储事件数据
- **事件索引**：建立事件索引
- **事件查询**：支持事件查询功能
- **接入报告**：生成事件接入报告

---

## 使用示例

### 输出示例
```json
{
  "source_info": {
    "source_type": "message_queue",
    "source_name": "kafka_cluster",
    "topics": ["news", "announcements", "tickets"],
    "consumer_group": "event_ingestion_group"
  },
  "ingestion_config": {
    "subscription_mode": "subscribe",
    "filter_rules": {
      "event_type": ["news", "announcement"],
      "priority": ["high", "medium"]
    },
    "batch_size": 100,
    "processing_mode": "real_time"
  },
  "ingestion_results": {
    "total_events": 10000,
    "processed_events": 9800,
    "failed_events": 200,
    "ingestion_time": "2024-03-15T10:00:00",
    "duration": "3600s"
  },
  "event_samples": [
    {
      "event_id": "EVT001",
      "event_type": "news",
      "event_topic": "news",
      "event_time": "2024-03-15T10:00:00",
      "event_data": {
        "title": "市场动态",
        "content": "事件内容...",
        "source": "财经新闻"
      },
      "metadata": {
        "priority": "high",
        "category": "market"
      },
      "processed": true
    }
  ],
  "statistics": {
    "events_received": 10000,
    "events_processed": 9800,
    "events_stored": 9800,
    "processing_rate": "2.7 events/s",
    "success_rate": 0.98
  }
}
```

---

## 注意事项与限制

### 1. 事件流要求
- 事件流需要稳定可靠
- 连接信息需要准确
- 订阅配置需要正确

### 2. 事件处理性能
- 高频事件可能影响性能
- 需要合理设置批处理大小
- 需要优化处理流程

### 3. 事件数据质量
- 事件数据质量影响处理结果
- 需要验证事件完整性
- 异常事件需要处理

### 4. 事件去重
- 需要识别重复事件
- 去重策略需要选择
- 可能影响事件顺序

### 5. 使用限制
- 本 Skill 不包含事件源管理功能
- 接入结果需要人工复核
- 复杂事件流可能需要特殊处理

---

## 参考资料
- 见 references/ 目录中的相关文档，包括：
  - 事件流接入方法手册
  - 消息队列配置指南
  - 事件处理策略说明
  - 性能优化指南