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. 高级处理功能

• 事件存储：存储事件数据
• 事件索引：建立事件索引
• 事件查询：支持事件查询功能
• 接入报告：生成事件接入报告

---

使用示例

输出示例

{
  "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/ 目录中的相关文档，包括：
  • 事件流接入方法手册
  • 消息队列配置指南
  • 事件处理策略说明
  • 性能优化指南