ByteHouse（云数仓版）支持通过 Kafka 流式传输数据。本文档介绍如何将日志服务中的日志数据通过 Kafka 协议消费到 ByteHouse。

背景信息

日志服务支持通过 Kafka 协议消费指定日志主题中的日志数据，例如消费到 ByteHouse（云数仓版）中进行进一步的分析处理。在 ByteHouse 中创建 Kafka 数据导入任务之后，可以直接通过 Kafka 流式传输数据。数据导入任务将自动运行，持续读取日志主题中的日志数据，并将其写入到指定的数据库表中。
消费日志时，支持仅消费其中的部分字段，并设置最大消息大小等配置。同时您可以随时停止数据导入任务以减少资源使用，并在任何必要的时候恢复该任务。ByteHouse 将在内部记录 offset，以确保停止和恢复过程中不会丢失数据。

费用说明

通过 ByteHouse 消费日志时，涉及日志服务读流量费用。推荐使用私网服务地址，通过私网消费日志数据，对应的内网读流量费用请参考计费项与价格。
日志消费到 ByteHouse 之后，在 ByteHouse 侧会产生存储费用。如果针对这些数据进行实时查询分析，还会产生一定的计算费用。详细的价格请参考计费概述。

前提条件

• 已开通日志服务，创建日志项目与日志主题，并成功采集到日志数据。详细说明请参考快速入门。
• 已为指定日志主题开启 Kafka 协议消费功能，开启方式请参考开启 Kafka 消费功能。
• 推荐使用 IAM 账号进行操作。操作前，需确保当前登录账号拥有 Kafka 协议消费的权限，即具备 Action ConsumeLogs 的权限。详细信息请参考可授权的操作。
• 已开通 ByteHouse（云数仓版）。操作步骤请参考开通服务。

注意事项

通过 ByteHouse 消费日志时，最大可消费的日志大小为 128KB，超过限制的日志数据将无法被成功消费到 ByteHouse。该限制由数据导入任务的 Max Block Size 配置指定，该配置的取值范围为 65,536~131,072 字节，即最大 128KB。

导入日志到 ByteHouse

ByteHouse 控制台提供新版和旧版数据加载界面，本文档以旧版数据加载控制台界面为例演示新建 Kafka 数据源和新建数据导入任务步骤。

1. 登录 ByteHouse 控制台。
2. 新建数据库。

   1. 在数据库页面的左上角单击 + 新建数据库。

   2. 填写数据库的基本信息。

      配置	说明
      数据库名称	数据库的名称，只能包含字母、数字和下划线（_）。
      数据库描述	数据库的描述信息。

   3. 单击创建。

3. 新建数据表。
   1. 在指定数据库的详情页面中单击进入数据表页签。
   2. 在页面右上角单击 + 数据表。
   3. 选择SQL或可视化页面创建。
   4. 在 SQL 或可视化页签中设置列名和数据类型。
      您需要根据日志服务中源日志结构设计新的表结构，建议仅创建需要保存或用于后续分析的列。
   5. 单击创建。
4. 新建 Kafka 数据源。

   1. 在顶部导航栏单击数据加载。

   2. 在页面左上角单击数据源。

   3. 在数据源页面中，单击 + 连接新源。

   4. 选择数据源类型为 Kafka 数据流。

   5. 填写 Kafka 源的基本信息。

      配置	说明
      源名称	Kafka 数据源的名称。
      Kafka 代理列表 IP 地址	初始连接的集群地址。格式为服务地址:端口号，例如 tls-cn-beijing.ivolces.com:9093，其中： 服务地址为当前地域下日志服务的服务地址。请根据地域和网络类型选择正确的服务入口，详细信息请参见服务地址。推荐使用私网服务地址，节省流量费用。 端口号固定为 9093。
      身份验证模式	身份验证机制。此处应维持默认设置 PLAIN。
      SSL	是否开启 SSL。此处应维持默认的开启状态。
      用户名	应配置为日志服务的日志项目 ID。
      密码	应配置为火山引擎账户密钥。格式为 ${access-key-id}#${access-key-secret}，其中： ${access-key-id} 应替换为您的 AccessKey ID。 ${access-key-secret} 应替换为您的 AccessKey Secret。 说明 如果当前登录用户为 IAM 用户，则此处应配置为 IAM 用户的密钥。

   6. 单击连接。

5. 新建数据导入任务。

   1. 在顶部导航栏单击数据加载。

   2. 在页面左上角单击 + 新建数据加载。

   3. 填写基本信息。

      配置	说明
      任务名称	数据导入任务的名称。长度为 1~128 个字符。
      任务描述	数据导入任务的描述信息。

   4. 选择数据源。

      配置	说明
      分类	选择 Kafka 数据流。
      数据源	选择步骤 4 中创建的 Kafka 数据源。
      Topic	Kafka 协议消费主题 ID，格式为 out-日志主题ID，例如 out-0fdaa6b6-3c9f-424c-8664-fc0d222c****。 您也可以在日志服务控制台的 Topic 详情页中查看并复制 Kafka 协议消费主题 ID。
      消费者组	Kafka 消费组，用于消费指定日志主题中的日志数据。 如果从未创建过，可以单击新建消费者组临时创建一个。
      格式	数据格式。此处应固定为 JSON_KAFKA。

   5. 选择目标表。

      配置	说明
      目标数据库	日志服务日志数据在 ByteHouse 中存储的数据库名称。
      目标表	日志服务日志数据在 ByteHouse 中存储的表名称。

   6. 定义 Schema 映射。 设置日志服务中日志数据和 ByteHouse 中数据表的字段映射关系。指定映射关系之后，指定字段在 ByteHouse 中将以对应的列名及数据类型进行存储。

      说明

      • 如果日志主题中已有日志数据，您可以直接单击解析来自动解析数据。
      • 如果表结构不符合需求，也可以单击编辑目标表来修改目标表的结构。
      • 字段的源列名称和数据类型必须与源日志数据中的字段完全匹配，否则该字段无法成功导入。

   7. 设置导入任务配置。

      配置	说明
      加载类型	数据加载的方式。当前仅支持增量写入，即在每次执行作业中，日志数据将被增量写入到目标表，存量数据不会被覆盖。
      Max Block Size	最大消息大小。单位为字节，取值范围为 65,536~131,072 字节，即最大可设置为 128 KB。

   8. 单击创建。 成功创建数据导入任务之后，该任务将自动运行，持续通过 Kafka 协议流式消费日志服务中的日志数据。如果需要暂停数据导入任务，可以在导入任务的详情页右上角单击停止。

查看导入任务

执行计划 > 实时数据加载页面中展示当前已创建的实时数据导入任务的运行信息。单击导入任务的名称即可查看详细信息。

同步状态说明

数据导入任务的同步状态说明如下。

常见问题

新建 Kafka 数据源时报错无法连接到 Broker

• 问题现象：在步骤 4 新建 Kafka 数据源时，页面报错 “kafka connectivity test failed，ping failed with err: kafka: client has run out of available brokers to talk to.......”。
• 可能原因：日志主题未开启 Kafka 协议消费、或 Kafka 代理列表 IP 地址、用户名及密码填写错误。
• 解决方案：
  • 为指定日志主题开启 Kafka 协议消费功能，开启方式请参考开启 Kafka 消费功能。
  • 参考操作步骤正确填写相关参数，操作步骤说明请参考操作步骤。