一、前提条件
·已 开通腾讯云消息队列 （CKafka）。
·确保当前操作账号拥有开通投递到 CKafka 的权限。详情请参见 CLS 访问策略模板。
二、操作步骤
1. 准备和配置 CKafka Topic。
1.1 在日志主题同地域下，创建一个 CKafka 实例，详情请参见 创建实例。
1.2 创建一个 Topic，详情请参见 创建 Topic。进入该 Topic 的高级配置，修改如下配置，其余参数可按需配置。
·cleanup.policy：选择 delete，否则会投递失败。
·max.message.bytes：设置为12MB。
2. 进入投递 CKafka 的任务管理页面，首先登录 日志服务控制台，有两种方式进入投递任务管理:
·入口1：在左侧导航栏中，选择投递消费 > 投递到 CKafka，然后在页面顶部，依次选择地域、日志集和日志主题。
·入口2：在日志主题列表，选择并进入需要投递 CKafka 的日志主题，在日志主题详情页面，选择页面左侧的投递消费 > 投递到 CKafka。
3. 单击右侧的编辑，开启投递到 CKafka 开关，选择相应的 CKafka 实例以及对应的 Topic，开始投递。你可以选择以原始内容或者 JSON 投递日志。
·以原始内容投递到 CKafka。配置项说明如下：
| 配置项 | 解释说明 | 规则 | 是否必填 |
| 目标 CKafka Topic 归属 | 当前主账号其他主账号投递 CLS 日志至当前主账号的 CKafka Topic。投递 CLS 日志至其他主账号的 CKafka，例如 A 账号在 CLS 将日志投递至 B 账号的 CKafka Topic，需要 B 账号在 CAM(访问管理)侧配置访问 角色，配置完成之后，由 A 账号将角色 ARN 和外部 ID 填写到 CLS 控制台，方可进行跨账号投递。配置角色的步骤如下：1. 新建角色。账号 B 登录 CAM 角色管理页面。1.1 创建访问策略，策略名称例如：cross_shipper，策略语法参考如下：说明：示例中的授权按照最小权限的原则，resource 配置为仅可以投递至广州地域的 CKafka：ckafka-12abcde3，请你按照实际情况进行授权。 { "statement": [ { "action": [ "cam:GetRole" ], "effect": "allow", "resource": [ "*" ] }, { "action": [ "ckafka:ListRoute", "ckafka:AddRoute", "ckafka:DescribeInstanceAttributes", "ckafka:AuthorizeToken", "ckafka:CreateToken", "ckafka:DescribeTopicAttributes" ], "effect": "allow", "resource": [ "qcs::ckafka:ap-guangzhou:uin/100001234567:ckafkaId/ckafka-12abcde3" ]//resource 配置为仅可以投递至广州地域的 CKafka：ckafka-12abcde3，请你按照实际情况进行授权 } ], "version": "2.0"}1.2 新建角色，选择腾讯云账户角色载体，云账号类型选择其他主账号，然后输入 A 账号的 ID，例如100012345678，勾选开启校验并配置外部 ID，例如：Hello123。1.3 配置角色策略，配置角色的访问策略，选择第一步配置好的访问策略 cross_shipper(示例)。1.4 保存该角色，例如：uinA_writeCLS_to_CKafka。2. 为角色配置载体。在 CAM 角色列表中找到 uinA_writeCLS_to_CKafka(示例)，单击该角色，选中下方的角色载体 > 管理载体 > 添加产品服务 > 选中日志服务，然后单击更新。可以看到当前角色的载体是两个：一个是 A 账号，另一个是 cls.cloud.tencent.com(CLS 日志服务)。3. A 账号登录 CLS，填入角色 ARN 和外部 ID。该两项信息需 B 账号来提供：B 账号在 CAM 角色列表中找到角色 uinA_writeCLS_to_CKafka(示例)，单击可查看该角色的 RoleArn，例如 qcs::cam::uin/100001112345:roleName/uinA_writeCLS_to_CKafka。在角色载体中可看到外部 ID，例如 Hello123。注意：填写角色 ARN，外部 ID 时，注意不要输入多余的空格，会导致权限校验失败。跨账号投递会在 A 账号下，产生日志主题的读流量费。 | 当前主账号/其他主账号 | 否 |
| CKafka 实例 | 与当前日志主题同地域的 CKafka Topic 作为投递目标。跨账号投递场景下，需用户手动填写 CKafka 实例 ID 和 Topic 名称。 | 列表选择 | 必填 |
| 投递数据格式 | 选择原始内容，投递用户的原始日志。 | 列表选择 | 必填 |
| 数据压缩格式 | 不压缩\\SNAPPY\\LZ4。 | 列表选择 | 必填 |
| 投递日志预览 | 预览你投递的日志数据。 | – | – |
·以 JSON 格式投递至 CKafka。配置项说明如下：
| 配置项 | 解释说明 | 规则 | 是否必填 |
| 目标 CKafka Topic 归属 | 当前主账号其他主账号投递 CLS 日志至当前主账号的 CKafka Topic。投递 CLS 日志至其他主账号的 CKafka，例如 A 账号在 CLS 将日志投递至 B 账号的 CKafka Topic，需要 B 账号在 CAM(访问管理)侧配置访问 角色，配置完成之后，由 A 账号将角色 ARN 和外部 ID 填写到 CLS 控制台，方可进行跨账号投递。配置角色的步骤如下：1. 新建角色。账号 B 登录 CAM 角色管理页面。1.1 创建访问策略，策略名称例如：cross_shipper，策略语法参考如下：说明：示例中的授权按照最小权限的原则，resource 配置为仅可以投递至广州地域的 CKafka：ckafka-12abcde3，请你按照实际情况进行授权。 { "statement": [ { "action": [ "cam:GetRole" ], "effect": "allow", "resource": [ "*" ] }, { "action": [ "ckafka:ListRoute", "ckafka:AddRoute", "ckafka:DescribeInstanceAttributes", "ckafka:AuthorizeToken", "ckafka:CreateToken", "ckafka:DescribeTopicAttributes" ], "effect": "allow", "resource": [ "qcs::ckafka:ap-guangzhou:uin/100001234567:ckafkaId/ckafka-12abcde3" ]//resource 配置为仅可以投递至广州地域的 CKafka：ckafka-12abcde3，请你按照实际情况进行授权 } ], "version": "2.0"}1.2 新建角色，选择腾讯云账户角色载体，云账号类型选择其他主账号，然后输入 A 账号的 ID，例如100012345678，勾选开启校验并配置外部 ID，例如：Hello123。1.3 配置角色策略，配置角色的访问策略，选择第一步配置好的访问策略 cross_shipper(示例)。1.4 保存该角色，例如：uinA_writeCLS_to_CKafka。2. 为角色配置载体。在 CAM 角色列表中找到 uinA_writeCLS_to_CKafka(示例)，单击该角色，选中下方的角色载体 > 管理载体 > 添加产品服务 > 选中日志服务，然后单击更新。可以看到当前角色的载体是两个：一个是 A 账号，另一个是 cls.cloud.tencent.com(CLS 日志服务)。3. A 账号登录 CLS，填入角色 ARN 和外部 ID。该两项信息需 B 账号来提供：B 账号在 CAM 角色列表中找到角色 uinA_writeCLS_to_CKafka(示例)，单击可查看该角色的 RoleArn，例如 qcs::cam::uin/100001112345:roleName/uinA_writeCLS_to_CKafka。在角色载体中可看到外部 ID，例如 Hello123。注意：填写角色 ARN，外部 ID 时，注意不要输入多余的空格，会导致权限校验失败。跨账号投递会在 A 账号下，产生日志主题的读流量费。 | 当前主账号/其他主账号 | 否 |
| CKafka 实例 | 与当前日志主题同地域的 CKafka Topic 作为投递目标。 | 列表选择 | 必填 |
| 投递数据格式 | 选择 JSON，以 JSON 的数据格式投递日志。 | 列表选择 | 必填 |
| 投递日志字段 | 请选择你需要投递的日志字段, JSON 格式的转义/不转义的说明如下: 转义，将 JSON 第一层节点的值转为 String，如果你的第一层节点的值是 Struct，在下游入库或者计算时，需要提前将该 Struct 转为 String，可以选这个选项。不转义，不对你的 JSON 结构和层级做修改，日志格式和采集侧保持一致。注意：当 JSON 的第一层节点中包含有数值时，消费后会自动转为 int、float。日志：{"a":123, "b":"123", "c":"-123", "d":"123.45", "e":{"e1":123,"f1":"123"}}消费：{"a":123,"b":123,"c":-123,"d":123.45,"e":{"e1":123,"f1":"123"}}将__TAG__元信息平铺或者不平铺，说明如下：示例：__TAG__元信息：{"__TAG__":{"fieldA":200, "fieldB":"text"}}平铺：{"__TAG__.fieldA":200,"__TAG__.fieldB":"text"}不平铺：{"__TAG__":{"fieldA":200, "fieldB":"text"}} | 选择 | 必填 |
| 数据压缩格式 | 不压缩\\SNAPPY\\LZ4。 | 选择 | 必填 |
| 投递日志预览 | 预览你投递的日志数据。 | – | – |
| 高级选项 | CKafka 分区生产规则：打开开关后，将依据你配置的字段值，计算 Hash 值，结果相同的，会写入 CKafka 的同一个分区。注意：最多可使用5个字段的值进行拼接，多个字段值的字符总长度不超过4096，超出会被截断。尽量选择字段值为字符串或者数值的进行 Hash，字段值为复杂类型（如 JSON） 时，不保证按照 Hash 结果投递分区的准确性。 | 选择 | 必填 |
4. 单击确定，启动投递到 CKafka。
注意：如需在投递至 CKafka 前对日志进行清洗加工过滤，请参见使用 数据加工 操作。
三、提示没有读写 CKafka Topic 的权限，怎么办？
如果你直接使用 API 接口投递数据到 CKafka，可能会存在读写 CKafka Topic 的权限问题。因为，如果你在控制台使用该功能，系统会引导你完成相关授权；如果你直接调用 API 投递，则需要手动授权。具体的排查和解决方案请参见 投递任务角色授权。