云服务器特惠 99/年起
  1. 腾讯云优惠网首页
  2. 腾讯云产品

消息队列 CKafka 版 数据处理规则解析

• 腾讯云产品

在通过 CKafka 连接器处理数据流入流出任务时，通常需要对数据进行简易的清洗操作，如格式化原始数据、解析特定字段、数据格式转换等等。开发者往往需要自己搭建一套数据清洗的服务（ETL）。

相关产品
消息队列 CKafka 版 产品页，含版本规格、价格与购买入口。

一、概览

Logstash 是一款免费且开放的服务器端数据处理管道，能够从多个数据源采集数据，转换数据，然后将数据发送到相应的“存储库”中。 logstash 拥有丰富的过滤器插件，这使得 logstash 成为了被广泛使用的一款功能强大的数据转换工具。

然而搭建、配置、维护自己的 logstash 服务会增大开发和运维的难度，为此 CKafka 提供了一套对标 logstash 的数据处理服务，开发者仅需通过控制台交互界面就可以新建自己的数据处理任务。数据处理服务允许用户编辑相应的数据处理规则，支持构建链式处理，同时可以预览数据处理的结果，帮助用户轻松高效的构建一套数据处理服务，满足数据清洗和转换的需求。

二、功能对标清单

Logstash 连接器数据处理服务 功能
Codec.json ✔ 数据解析（JSON）
Filter.grok ✔ 数据解析（正则匹配）
Filter.mutate.split ✔ 数据解析（字符分割）
Filter.date ✔ 日期格式处理
Filter.json ✔ 解析内部 JSON 结构
Filter.mutate.convert ✔ 数据修改（格式转换）
Filter.mutate.gsub ✔ 数据修改（字符替换）
Filter.mutate.strip ✔ 数据修改（去除首尾空格）
Filter.mutate.join ✔ 数据修改（拼接字段）
Filter.mutate.rename ✔ 字段修改（更改字段名）
Filter.mutate.update ✔ 字段修改（更新字段）
Filter.mutate.replace ✔ 字段修改（替换字段）
Filter.mutate.add_field ✔ 字段修改（添加字段）
Filter.mutate.remove_field ✔ 字段修改（删除字段）
Filter.mutate.copy ✔ 字段修改（复制字段值）
Filter.mutate.merge  TODO
Filter.mutate.uppercase  TODO
Filter.mutate.lowercase  TODO

三、数据解析

通过选择相应的数据解析模式 ，并一键单击即可预览：

四、日期格式处理

1. 输入带日期格式的原始数据，以下为一个示例。

{ "@timestamp": "2022-02-26T22:25:33.210Z", "beat": { "hostname": "test-server", "ip": "6.6.6.6", "version": "5.6.9" }, "input_type": "log", "message": "{\\"userId\\":888,\\"userName\\":\\"testUser\\"}", "offset": 3030131}

2. 解析结果如下：

3. CKafka 连接器处理方式：

3.1 通过预设系统当前时间给某字段赋值

3.2 在数据处理模块通过 处理 value 功能来对日期数据进行处理。

处理模式选择转换时间格式，选择好时间格式、时区和日期格式，并确认。

4. 单击测试，可以看到转换后的时间格式。

五、解析内部 JSON 结构

1. 输入带嵌套 JSON 格式的原始数据，以下为一个示例。

{ "@timestamp": "2022-02-26T22:25:33.210Z", "beat": { "hostname": "test-server", "ip": "6.6.6.6", "version": "5.6.9" }, "input_type": "log", "message": "{\\"userId\\":888,\\"userName\\":\\"testUser\\"}", "offset": 3030131}

2. 解析结果如下：

3. 通过对该字段选择 MAP 操作来对其进行解析，从而把特定字段解析为 JSON 格式：

六、数据修改

输入原始数据，以下为一个示例。

{ "@timestamp": "2022-02-26T22:25:33.210Z", "beat": { "hostname": "test-server", "ip": "6.6.6.6", "version": "5.6.9" }, "input_type": "log", "message": "{\\"userId\\":888,\\"userName\\":\\"testUser\\"}", "offset": 3030131}

解析结果如下：

连接器处理方式如下：

·方式一：通过选择相应的处理 value 功能一键定义规则。

·方式二：通过选择数据类型一键更改相应字段的数据格式。

·更改前：

·更改后：

·方式三：通过 JSONPath 语法实现 join 的拼接功能。如使用 $.concat($.data.Response.SubnetSet[0].VpcId, \\"#\\", $.data.Response.SubnetSet[0].SubnetId, \\"#\\", $.data.Response.SubnetSet[0].CidrBlock) 语法拼接 Vpc 和子网的属性，并且通过 # 字符加以分割。关于 JSONPath 语法的详细介绍请参见 JsonPath 说明。

七、字段修改

在通过 CKafka 连接器进行数据处理的过程中，可以使用多种方式对解析后的数据字段进行编辑修改，以获得你理想的数据。如：

·在 KEY 栏可以对字段名称进行修改。

·在 VALUE 栏可以选择复制某字段的值。

·在下方单击添加可以新增字段。

·在右侧单击按钮可以删除字段。

八、案例1：多级字段解析

·输入 message：

{ "@timestamp": "2022-02-26T22:25:33.210Z", "beat": { "hostname": "test-server", "ip": "6.6.6.6", "version": "5.6.9" }, "input_type": "log", "message": "{\\"userId\\":888,\\"userName\\":\\"testUser\\"}", "offset": 3030131,}

·目标 message ：

{ "@timestamp": "2022-02-26T22:25:33.210Z", "input_type": "log", "hostname": "test-server", "ip": "6.6.6.6", "userId": 888, "userName": "testUser"}

·连接器配置方法：

1.1 处理链 1 配置如下：

1.2 处理链 1 结果如下：

{ "@timestamp": "2022-02-26T22:25:33.210Z", "input_type": "log", "message": "{\\"userId\\":888,\\"userName\\":\\"testUser\\"}", "hostname": "test-server", "ip": "6.6.6.6"}

1.3 处理链 2 配置如下：

1.4 处理链 2 结果如下：

{ "@timestamp": "2022-02-26T22:25:33.210Z", "input_type": "log", "hostname": "test-server", "ip": "6.6.6.6", "userId": 888, "userName": "testUser"}

九、案例2：非 JSON 数据解析

·输入 message ：

region=Shanghai$area=a1$server=6.6.6.6$user=testUser$timeStamp=2022-02-26T22:25:33.210Z

·目标 message：

{ "region": "Shanghai", "area": "a1", "server": "6.6.6.6", "user": "testUser", "timeStamp": "2022-02-27 06:25:33", "processTimeStamp": "2022-06-27 11:14:49"}

·连接器配置方法：

1.1 使用分隔符 $ 对原始 message 进行解析

1.2 初步解析结果：

{ "0": "region=Shanghai", "1": "area=a1", "2": "server=6.6.6.6", "3": "user=testUser", "4": "timeStamp=2022-02-26T22:25:33.210Z"}

1.3 使用分隔符 = 对结果二次解析：

1.4 二次解析结果：

{ "0": "region=Shanghai", "1": "area=a1", "2": "server=6.6.6.6", "3": "user=testUser", "4": "timeStamp=2022-02-26T22:25:33.210Z", "0.region": "Shanghai", "1.area": "a1", "2.server": "6.6.6.6", "3.user": "testUser", "4.timeStamp": "2022-02-26T22:25:33.210Z"}

1.5 对字段进行编辑、删减，调整时间戳格式，并新增当前系统时间字段：

·最终结果：

{ "region": "Shanghai", "area": "a1", "server": "6.6.6.6", "user": "testUser", "timeStamp": "2022-02-27 06:25:33", "processTimeStamp": "2022-06-27 11:14:49"}

相关内容：

  1. 消息队列 CKafka 版 正则提取解析模式说明 Ckafka 连接器的数据处理功能提供了根据正则表达式提取消息内容的能力，正则提取采用的是开源的正则提取包 re2 。Java 的标准正则表达式包 java.util.regex 以及其他被广泛使用的正则表达式包如 PCRE、Perl RE 和 Python（re），都使用回溯实现策略，即当一个 pattern 出现两个替代方案a|b 的时候，引擎将首先尝试匹配子模式a，如果匹配失败，它将重置输入流并尝试匹配子模式 。...
  2. 消息队列 CKafka 版 Filebeat 接入 CKafka Beats 平台 集合了多种单一用途数据采集器。这些采集器安装后可用作轻量型代理，从成百上千或成千上万台机器向目标发送采集数据。  Beats 有多种采集器，可以以根据自身的需求下载对应的采集器。这篇以 Filebeat（轻量型日志采集器）为例，向你介绍 Filebeat 接入 CKafka 的操作方法，及接入后常。...
  3. 消息队列 CKafka 版 Logstash 接入 CKafka Logstash 是一个开源的日志处理工具，可以从多个源头收集数据、过滤收集的数据并对数据进行存储作为其他用途。Logstash 灵活性强，拥有强大的语法分析功能，插件丰富，支持多种输入和输出源。Logstash 作为水平可伸缩的数据管道，与 Elasticsearch 和 Kibana 配合，在日志收集检索方面功能强大。...
  4. 消息队列 CKafka 版 Schema Registry 接入 CKafka 无论是使用传统的 Avro API 自定义序列化类与反序列化类，还是使用 Twitter 的 Bijection 类库实现 Avro 的序列化与反序列化，两种方法有相同的缺点：在每条 Kafka 记录里都嵌入了 Schema，从而导致记录的大小成倍增加。但是不管怎样，在读取记录时仍然需要用到整个 Schema，所以要先找到 Schema。 CKafka 提供了数据共用一个 Schema 的方法：将 Schema 中的。...
  5. 消息队列 CKafka 版 CKafka 到 ClickHouse 使用云数据仓库 TCHouse-C 方案 简述：在 clickhouse 平台创建 kafka 引擎表和物化视图，把 kafka 的表数据导入到数据仓库。...
  6. 消息队列 CKafka 版 对 CKafka 进行生产和消费压力测试 Kafka Producer 和 Consumer 的性能测试均可使用 Kafka 客户端自带的开源脚本，主要输出每秒发送消息量（MB/second）和每秒发送消息数（records/second）两项指标。...
  7. 消息队列 CKafka 版 连接器替代方案 腾讯云CKafka连接器更新日志（2021-2025）展示了数据接入、处理、流出及跨云同步等核心功能的持续迭代，支持K2K跨云同步、MySQL/PostgreSQL数据订阅优化、数据流出至COS/ES/CLS等目标，并新增连接管理、任务编排、事件中心等能力。 数据同步与跨云：K2K支持跨云及自建Kafka同步至CKafka，并支持同步消费位点。 数据订阅优化：MySQL/MariaDB订阅新增锁方式配置、正则匹配库表。...
  8. 消息队列 CKafka 版 Flume 接入 CKafka Apache Flume 是一个分布式、可靠、高可用的日志收集系统，支持各种各样的数据来源（如 HTTP、Log 文件、JMS、监听端口数据等），能将这些数据源的海量日志数据进行高效收集、聚合、移动，最后存储到指定存储系统中（如 Kafka、分布式文件系统、Solr 搜索服务器等）。...
赞 (0)
上一篇 4天前
下一篇 4天前