odpf/beast

Beast

镜像概述和主要用途

Beast是一个Kafka到BigQuery的接收器（Sink）工具，主要用于将Kafka集群中的消息批量、可靠地推送到BigQuery。它通过消费者、BigQuery工作器和提交器等组件协同工作，确保消息从Kafka消费、处理到BigQuery存储的完整流程，并支持无效消息的死信处理，适用于流数据持久化场景。

核心功能和特性### 架构组件

• 消费者（Consumer）：从Kafka批量消费消息，将消息推送到读取队列和提交队列（均为阻塞队列）。当队列满时会停止消费（可通过轮询超时配置调整）。
• BigQuery工作器（BigQuery Worker）：从读取队列轮询消息并推送到BigQuery。推送成功后，向提交器发送确认。
• 提交器（Committer）：接收BigQuery工作器的成功推送确认并存储在集合中，同时从提交队列轮询消息批次。若批次存在于确认集合中（即已成功推送至BigQuery），则提交该批次的最大偏移量到Kafka，并从队列和集合中移除该批次。
• 死信处理（Dead Letters）：提供可插拔的GCS（Google Cloud Storage）组件，用于存储被BigQuery拒绝的无效消息（如分区键为时间戳但超出范围的消息）。默认情况下，遇到无效消息会停止处理，可通过环境变量启用GCS存储：

  ENABLE_GCS_ERROR_SINK=true
  GCS_BUCKET=<谷歌云存储桶名称>
  GCS_PATH_PREFIX=<桶内路径前缀>
  GCS_WRITER_PROJECT_NAME=<包含存储桶的谷歌项目>
  

  无效消息在GCS中的存储路径格式为：<GCS_WRITER_PROJECT_NAME>/<GCS_BUCKET>/<GCS_PATH_PREFIX>/<dt=yyyy-MM-dd>/<topicName>/<random-uuid>，其中<topicName>为消息来源主题，<random-uuid>为文件名。

使用场景和适用范围

适用于需要将Kafka流数据（如系统日志、业务事件、用户行为等）持久化到BigQuery进行后续分析的场景。确保数据从Kafka到BigQuery的可靠传输，避免数据丢失，并处理异常消息。

使用方法和配置说明### 前提条件

• 运行中的Kafka集群，消息需为proto格式
• 具有流写入权限的BigQuery项目
• 已为消息proto创建对应的BigQuery表
• 已配置表的列映射及环境变量（包含BigQuery、Kafka和应用参数）

本地运行

bash
git clone https://github.com/odpf/beast
export $(cat ./env/sample.properties | xargs -L1) && gradle clean runConsumer

Docker运行

镜像托管于https://hub.docker.com/r/odpf/beast%EF%BC%9A

bash
export TAG=release-0.1.1
docker run --env-file beast.env -v ./local_dir/project-secret.json:/var/bq-secret.json -it odpf/beast:$TAG

• -v：挂载本地密钥文件project-secret.json到容器内/var/bq-secret.json，环境变量GOOGLE_CREDENTIALS需指向该路径用于BigQuery认证
• TAG：可更新为最新镜像标签，示例标签已通过测试

Kubernetes部署

为Kafka中的目标主题创建Beast部署：

• 部署可包含多个Beast实例
• 单个Beast容器包含以下线程：Kafka消费者、多个BigQuery工作器、提交器
• 部署还包含Telegraf容器用于推送统计指标

Helm部署请参考https://github.com/gojektech/charts/tree/master/incubator/beast%E3%80%82

BigQuery设置

以TestMessage proto文件为例，使用schema文件创建表：

bash
# 从schema创建新表
bq mk --table <project_name>:dataset_name.test_messages ./docs/test_messages.schema.json

# 查询总记录数
bq query --nouse_legacy_sql 'SELECT count(*) FROM `<project_name>:dataset_name.test_messages LIMIT 10'

# 从本地schema文件更新BigQuery表结构
bq update --format=prettyjson <project_name>:dataset_name.test_messages booking.schema

# 导出表结构到文件
bq show --schema --format=prettyjson <project_name>:dataset_name.test_messages > test_messages.schema.json

生成Kafka消息

可使用https://github.com/gojekfarm/sample-kafka-producer%E7%94%9F%E6%88%90TestMessage.proto%E6%A0%BC%E5%BC%8F%E7%9A%84%E6%B6%88%E6%81%AF%E5%B9%B6%E6%8E%A8%E9%80%81%E5%88%B0Kafka%E3%80%82

运行Stencil Server

• 执行脚本./run_descriptor_server.sh在build目录构建描述符，并启动Python服务器（默认端口8000）
• Stencil URL可配置为curl http://localhost:8000/messages.desc

贡献

• 可提交issue或澄清问题
• 可提交PR解决功能或问题
• 可协助完善文档

本地测试命令：

bash
git clone https://github.com/odpf/beast
export $(cat ./env/sample.properties | xargs -L1) && gradlew test