snowplow/transformer-kinesis

Snowplow Transformer for Kinesis 镜像文档

镜像概述

Snowplow Transformer for Kinesis是Snowplow数据处理生态系统的核心组件，专门设计用于处理和转换来自Amazon Kinesis流的事件数据。作为数据处理管道的中间层，它接收原始或半结构化事件数据，执行规范化、丰富和格式转换操作，输出结构化数据以支持后续分析、存储或进一步处理。该镜像为容器化部署提供支持，简化了在各类环境中的集成与扩展。

核心功能与特性

• Kinesis流原生集成：无缝对接Amazon Kinesis Data Streams，支持配置输入流（待处理数据）和输出流（转换后数据）。
• 事件数据转换：支持将Snowplow原始事件数据（如Thrift、JSON格式）转换为结构化格式（如JSON、Parquet、Avro），适配下游数据系统需求。
• 数据规范化：统一事件字段格式，处理数据类型转换、字段映射及缺失值处理，确保输出数据一致性。
• 数据丰富能力：集成Snowplow Enrichment框架，支持通过配置加载丰富规则（如UA解析、地理位置解析、IP到ISP映射等），增强事件上下文信息。
• 容错与可靠性：内置错误处理机制，支持失败记录重试、死信队列（DLQ）配置，保障数据处理连续性。
• 灵活配置：通过环境变量或配置文件自定义转换规则、输出格式、丰富策略及资源限制，适配多样化业务场景。

使用场景与适用范围

• 实时数据处理管道：构建从Kinesis流到数据仓库（如Redshift）、数据湖（如S3）的实时数据链路，支持分钟级数据可见性。
• 事件数据标准化：将异构事件源（如Web、移动、IoT）产生的原始数据转换为统一格式，简化下游分析（如用户行为分析、产品埋点分析）。
• Snowplow生态集成：作为Snowplow完整数据管道（Collector→Enrich→Transformer→Loader）的关键环节，与Snowplow Collector、Loader等组件协同工作。
• AWS云原生部署：在AWS ECS、EKS或EC2环境中部署，结合IAM权限管理、CloudWatch监控，构建云原生数据处理架构。

使用方法与配置说明

前提条件

• 已创建Amazon Kinesis Data Stream（输入流），且流中包含Snowplow事件数据（需符合Snowplow事件格式规范）。
• 已配置目标存储（如Kinesis输出流、S3桶），并确保Transformer具备读写权限。
• （可选）如需启用数据丰富，需准备Enrichment配置文件（遵循Iglu schema规范）。

基础部署示例（Docker Run）

bash
docker run -d \
  --name snowplow-transformer-kinesis \
  -e AWS_REGION="us-east-1" \
  -e INPUT_STREAM_NAME="snowplow-raw-events" \
  -e OUTPUT_STREAM_NAME="snowplow-transformed-events" \
  -e TRANSFORMATION_CONFIG_PATH="/config/transformer.json" \
  -e LOG_LEVEL="INFO" \
  -v /host/path/to/config:/config \
  snowplow/transformer-kinesis:latest

核心配置参数

环境变量配置

*注：若无需输出到Kinesis流，可配置OUTPUT_S3_BUCKET指定S3目标存储（需额外配置S3相关参数）。

转换规则配置文件示例（transformer.json）

json
{
  "schema": "iglu:com.snowplowanalytics.snowplow/transformer_config/jsonschema/1-0-0",
  "data": {
    "outputFormat": "JSON",  // 输出格式：JSON/Parquet/Avro
    "compression": "NONE",   // 压缩方式：NONE/GZIP/SNAPPY
    "enrichments": [         // 启用的数据丰富规则（需提前加载至Iglu注册表）
      "iglu:com.snowplowanalytics.snowplow/ua_parser_enrichment/jsonschema/1-0-0",
      "iglu:com.snowplowanalytics.snowplow/geo_ip_enrichment/jsonschema/2-0-0"
    ],
    "validations": {
      "strictSchemaValidation": true,  // 是否启用严格Schema校验
      "allowMissingFields": false      // 是否允许缺失非必需字段
    },
    "fieldMappings": {       // 自定义字段映射（覆盖默认字段名）
      "app_id": "application_id",
      "event_name": "event_type"
    }
  }
}

进阶部署（Docker Compose）

yaml
version: "3.8"
services:
  transformer:
    image: snowplow/transformer-kinesis:latest
    container_name: snowplow-transformer
    environment:
      - AWS_REGION=us-east-1
      - INPUT_STREAM_NAME=snowplow-raw-events
      - OUTPUT_STREAM_NAME=snowplow-transformed-events
      - TRANSFORMATION_CONFIG_PATH=/config/transformer.json
      - LOG_LEVEL=INFO
      - WORKER_THREADS=8
    volumes:
      - ./config:/config  # 挂载本地配置目录
    restart: unless-stopped
    resources:
      limits:
        cpus: '2'
        memory: 4G

部署注意事项

• 资源配置：根据Kinesis流吞吐量（分片数）调整CPU/内存资源，建议每5个分片分配1核CPU及2GB内存。
• 权限管理：生产环境中建议通过IAM角色（如ECS任务角色、EKS服务账户）授予权限，避免硬编码AWS密钥。
• 监控与告警：集成CloudWatch或Prometheus监控容器指标（如处理延迟、成功/失败记录数），配置告警阈值（如失败率>1%触发告警）。
• 版本兼容性：确保与Snowplow Collector、Loader组件版本兼容（参考Snowplow官方版本矩阵）。
• 数据备份：建议对输入流启用Kinesis数据备份，或配置Transformer输出至S3归档，防止数据丢失。