Quickwit获取Kafka数据源消息

缠丝猫  金牌会员 | 9 小时前 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题
楼主

主题 887|帖子 887|积分 2661

先容

Quickwit可以将数据从一个或多个源插入到索引中。创建索引后，可以利用CLI 命令quickwit source create添加源，支持的源有：file、kafka、kinesis、pulsar。
本章解说如何从Quickwit搜刮引擎中创建Kafka源和获取Kafka源主题数据流，注意从Kafka流中读取数据流中的每条消息都必须包含一个JSON 对象，现在官方只支持数据源JSON格式数据导入。
官方关于Kafka源文档阐明：Kafka | Quickwit
创建Quickwit索引

因此前已在docker服务主机上创建了包含quickwit服务的docker容器，本文不详述quickwit安装过程，可参见《Docker安装Quickwit搜刮引擎》。
官方示例YMAL配置
  1. #
  2. # Index config file for gh-archive dataset.
  3. #
  4. version: 0.8
  6. index_id: gh-archive
  8. doc_mapping:
  9.   field_mappings:
  10.     - name: id
  11.       type: text
  12.       tokenizer: raw
  13.     - name: type
  14.       type: text
  15.       fast: true
  16.       tokenizer: raw
  17.     - name: public
  18.       type: bool
  19.       fast: true
  20.     - name: payload
  21.       type: json
  22.       tokenizer: default
  23.     - name: org
  24.       type: json
  25.       tokenizer: default
  26.     - name: repo
  27.       type: json
  28.       tokenizer: default
  29.     - name: actor
  30.       type: json
  31.       tokenizer: default
  32.     - name: other
  33.       type: json
  34.       tokenizer: default
  35.     - name: created_at
  36.       type: datetime
  37.       fast: true
  38.       input_formats:
  39.         - rfc3339
  40.       fast_precision: seconds
  41.   timestamp_field: created_at
  43. indexing_settings:
  44.   commit_timeout_secs: 10
复制代码
也可直接下载官方示例yaml文件，并通过quickwit服务命令创建索引
  1. # Download GH Archive index config.
  2. wget -O gh-archive.yaml https://raw.githubusercontent.com/quickwit-oss/quickwit/main/config/tutorials/gh-archive/index-config.yaml
  4. # Create index.
  5. ./quickwit index create --index-config gh-archive.yaml
复制代码
Kafka服务安装

可自行百度，非本文焦点，略过...

下载Kafka测试数据

测试数据来源于Quickwit官方，参见：Kafka | Quickwit
创建kafka主题，下载官方提供的json测试数据包，将数据推送到Kafka主题；
  1. # Create a topic named `gh-archive` with 3 partitions.
  2. bin/kafka-topics.sh --create --topic gh-archive --partitions 3 --bootstrap-server localhost:9092
  4. # Download a few GH Archive files.
  5. wget https://data.gharchive.org/2022-05-12-{10..15}.json.gz
  7. # Load the events into Kafka topic.
  8. gunzip -c 2022-05-12*.json.gz | \
  9. bin/kafka-console-producer.sh --topic gh-archive --bootstrap-server localhost:9092
复制代码
由于文件包太大，本地下载其中1天json数据包，实行命令如下
  1. curl https://data.gharchive.org/2022-05-12-10.json.gz -o 2022-05-12-10.json.gz
复制代码
创建Quickwit数据源

kafka数据源yaml配置示例
  1. #
  2. # Kafka source config file.
  3. #
  4. version: 0.8
  5. source_id: kafka-source
  6. source_type: kafka
  7. num_pipelines: 2
  8. params:
  9.   topic: gh-archive
  10.   client_params:
  11.     bootstrap.servers: localhost:9092
复制代码
也可直接下载官方提供的kafka-source.yaml文件，通过quickwit服务命令直接创建gh-archive索引对应的kafka源；
  1. # Download Kafka source config.
  2. wget https://raw.githubusercontent.com/quickwit-oss/quickwit/main/config/tutorials/gh-archive/kafka-source.yaml
  4. # Create source.
  5. ./quickwit source create --index gh-archive --source-config kafka-source.yaml
复制代码
完成上述操纵后，Quickwit会创建索引器和搜刮器，索引器将会边接到Kafka源指定topic主题上，并从对应group组主题分区上获取数据，通过流式传输到Quickwit中；
Java推送到Kafka主题

基于Java发送消息到Kafka示例代码，工具类可参见《Kafka消息服务之Java工具类》，本章不作详细讲述，通过代码片断进行演示；
  1. //生产者发送消息
  2. KafkaUtils.KafkaStreamServer kafkaStreamServer =  KafkaUtils.bulidServer().createKafkaStreamServer("192.168.1.3", 9092);
  3. String topic = "gh-archive";
  4. int n = 0;
  5. List<String>lines = FileUtils.readLines(new File("D:\\test\\kafka\\2022-05-12-10.json"), "UTF-8");
  6. for (String line : lines) {
  7.     System.out.println("发送消息：" + line.substring(0,30) + " ...");
  8.     //向kafka队列发送数据
  9.     kafkaStreamServer.sendMsg(topic, line);
  10.     if (n > 0 && n / 1000  == 0) {
  11.         //线程休眠
  12.         TimeUnit.MILLISECONDS.sleep(RandomUtils.nextInt(1, 200));
  13.     }
  14.     n ++;
  15. }
  16. //共累计推送到kafka数量:156040条
  17. System.out.println("共累计推送到kafka数量:" + n + "条");
  18. kafkaStreamServer.close();
复制代码
因官方测试数据比力大，本章节通过提前下载Json数据压缩包，只利用了其中一个Json压缩包：2022-05-12-10.json.gz；
  1. curl https://data.gharchive.org/2022-05-12-10.json.gz -o 2022-05-12-10.json.gz
复制代码
Java创建Quickwit索引和Kafka数据源

Quickwit提供了丰富的REST API，因此支持通过HTTP请求创建、维护索引、索引查询以及数据源维护等；以下直接利用Java步伐通过Http演示示例；
QuickwitKafkaSourceTest.java
  1. package com.example;
  3. import org.junit.jupiter.api.Test;
  5. public class QuickwitKafkaSourceTest {
  6.     private final static String QUICKWIT_URL = "http://192.168.1.3:7280/";
  7.     /**
  8.      * 创建索引
  9.      * @throws Exception
  10.      */
  11.     @Test
  12.     public void createIndex() throws Exception {
  13.         String indexConf = """
  14.                 {
  15.                     "version": "0.8",
  16.                     "index_id": "gh-archive",
  17.                     "doc_mapping": {
  18.                         "field_mappings": [
  19.                             {
  20.                                 "name": "id",
  21.                                 "type": "text",
  22.                                 "tokenizer": "raw"
  23.                             },
  24.                             {
  25.                                 "name": "type",
  26.                                 "type": "text",
  27.                                 "fast": true,
  28.                                 "tokenizer": "raw"
  29.                             },
  30.                             {
  31.                                 "name": "public",
  32.                                 "type": "bool",
  33.                                 "fast": true
  34.                             },
  35.                             {
  36.                                 "name": "payload",
  37.                                 "type": "json",
  38.                                 "tokenizer": "default"
  39.                             },
  40.                             {
  41.                                 "name": "org",
  42.                                 "type": "json",
  43.                                 "tokenizer": "default"
  44.                             },
  45.                             {
  46.                                 "name": "repo",
  47.                                 "type": "json",
  48.                                 "tokenizer": "default"
  49.                             },
  50.                             {
  51.                                 "name": "actor",
  52.                                 "type": "json",
  53.                                 "tokenizer": "default"
  54.                             },
  55.                             {
  56.                                 "name": "other",
  57.                                 "type": "json",
  58.                                 "tokenizer": "default"
  59.                             },
  60.                             {
  61.                                 "name": "created_at",
  62.                                 "type": "datetime",
  63.                                 "fast": true,
  64.                                 "input_formats": ["rfc3339"],
  65.                                 "fast_precision": "seconds"
  66.                             }
  67.                         ],
  68.                         "timestamp_field": "created_at"
  69.                     },
  70.                     "indexing_settings": {
  71.                         "commit_timeout_secs": 10
  72.                     }
  73.                 }
  74.                 """;
  75.         System.out.println("响应：" + HttpUtils.build().httpPost(QUICKWIT_URL + "api/v1/indexes", indexConf));
  76.     }
  78.     /**
  79.      * 创建数据源， 源支持：（kafka，kinesis，file）
  80.      * @throws Exception
  81.      */
  82.     @Test
  83.     public void createSources() throws Exception {
  84.         String indexId = "gh-archive";
  85.         String jsonDate = """
  86.                 {
  87.                     "version": "0.8",
  88.                     "source_id": "kafka-source",
  89.                     "source_type": "kafka",
  90.                     "num_pipelines": 1,
  91.                     "input_format": "json",
  92.                     "params": {
  93.                         "topic": "gh-archive",
  94.                         "client_params": {
  95.                             "auto.offset.reset": "earliest",
  96.                             "bootstrap.servers": "192.168.1.5:9092"
  97.                         }
  98.                     }
  99.                 }
  100.                 """;
  101.         System.out.println("响应：" + HttpUtils.build().httpPost(QUICKWIT_URL + "api/v1/indexes/" + indexId + "/sources", jsonDate));
  102.     }
  104.     /**
  105.      * 将索引切换到一个新的源上
  106.      * @throws Exception
  107.      */
  108.     @Test
  109.     public void indexesSourceToggle() throws Exception {
  110.         String indexId = "gh-archive";
  111.         String sourceId = "kafka-source";
  112.         System.out.println("响应：" + HttpUtils.build().http(QUICKWIT_URL + "api/v1/indexes/" + indexId + "/sources/" + sourceId + "/toggle", "PUT", ""));
  113.     }
  114.    
  115.     /**
  116.      * 重置索引绑定的源
  117.      * @throws Exception
  118.      */
  119.     @Test
  120.     public void indexesSourceReset() throws Exception {
  121.         String indexId = "gh-archive";
  122.         String sourceId = "kafka-source";
  123.         System.out.println("响应：" + HttpUtils.build().http(QUICKWIT_URL + "api/v1/indexes/" + indexId + "/sources/" + sourceId + "/reset-checkpoint", "PUT", ""));
  124.     }
  126.     /**
  127.      * 删除索引绑定的源
  128.      * @throws Exception
  129.      */
  130.     @Test
  131.     public void indexesSourceDelete() throws Exception {
  132.         String indexId = "gh-archive";
  133.         String sourceId = "my-kafka-source";
  134.         System.out.println("响应：" + HttpUtils.build().http(QUICKWIT_URL + "api/v1/indexes/" + indexId + "/sources/" + sourceId, "DELETE", null));
  135.     }
  137.     /**
  138.      * 获取索引描述信息
  139.      * @throws Exception
  140.      */
  141.     @Test
  142.     public void getIndexesDescribe() throws Exception {
  143.         String indexId = "gh-archive";
  144.         System.out.println("响应：" + HttpUtils.build().httpGet(QUICKWIT_URL + "api/v1/indexes/" + indexId + "/describe"));
  145.     }
  147.     /**
  148.      * 删除索引数据和所有元数据
  149.      * @throws Exception
  150.      */
  151.     @Test
  152.     public void deleteIndexes() throws Exception {
  153.         String indexId = "gh-archive";
  154.         System.out.println("响应：" + HttpUtils.build().http(QUICKWIT_URL + "api/v1/indexes/" + indexId , "DELETE", null));
  155.     }
  157. }
复制代码
HttpUtils.java
  1. package com.example;
  3. import java.net.URI;
  4. import java.net.http.HttpClient;
  5. import java.net.http.HttpRequest;
  6. import java.net.http.HttpResponse;
  7. import java.nio.charset.StandardCharsets;
  9. public class HttpUtils {
  10.     private HttpClient httpClient;
  12.     public static HttpUtils build(){
  13.         return new HttpUtils();
  14.     }
  16.     private HttpUtils() {
  17.         httpClient = HttpClient.newHttpClient();
  18.     }
  20.     public String httpPost(String url, String jsonData) throws Exception {
  21.         return http(url, "POST", jsonData);
  22.     }
  24.     public String httpGet(String url) throws Exception {
  25.         return http(url, "GET", null);
  26.     }
  28.     public String http(String url, String method, String jsonData) throws Exception {
  29.         System.out.println("请求：" + url);
  30.         HttpRequest request = null;
  31.         HttpRequest.Builder builder = HttpRequest.newBuilder()
  32.                 .uri(new URI( url))
  33.                 .header("Content-Type", "application/json; charset=UTF-8" )
  34.                 .header("Timeout", "5000");
  35.         if ("POST".equals(method)) {
  36.             request = builder.POST(HttpRequest.BodyPublishers.ofString(jsonData, StandardCharsets.UTF_8)).build();
  37.         } else if ("PUT".equals(method)) {
  38.             request = builder.PUT(HttpRequest.BodyPublishers.ofString(jsonData, StandardCharsets.UTF_8)).build();
  39.         } else if ("DELETE".equals(method)) {
  40.             request = builder.DELETE().build();
  41.         } else {
  42.             request = builder.GET().build();
  43.         }
  44.         HttpResponse<String> response = httpClient.send(request, HttpResponse.BodyHandlers.ofString());
  45.         System.out.println("状态码：" + response.statusCode());
  46.         return response.body();
  47.     }
  48. }
复制代码
在Java开发工具中分别实行createIndex()和createSources()测试方法后，即可创建索引、绑定Kafka数据源；
检索Kafka数据

在推送Kafka数据后，通过Redpanda控制台工具（此工具本文不做详述，参见官方《redpanda-console-kafka-ui》）查看Kfaka主题巨细约438M；而且Json数据格式中有多层json嵌套；

在浏览器上输入Quickwit UI访问地点：http://127.0.0.1:7280，在Quickwit UI 》 Indexs 中查看已创建的gh-archive索引；

点击gh-archive索引，进入到SOURCES中查看已配Kafka源；

Quickwit索引器从Kafka源中获取数据流后写入到指定索引中，在Query editor功能中可以查询到已存储的索引数据；

参考：
REST API | Quickwit

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

正序浏览
返回列表

快速回复

高级模式
B Color Image Link Quote Code Smilies
您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

缠丝猫

金牌会员
这个人很懒什么都没写!

楼主热帖

标签云

挺好的 服务器
微信订阅号
微信服务号
微信客服
小程序
H5
关于我们 商务合作 网站地图
©Copyright 2022-2024 杭州祥升科技有限公司 版权所有 浙ICP备20004199号
快速回复 返回顶部 返回列表