ElasticSearch

本文主要介绍在 Linkis 中，ElasticSearch 引擎连接器的安装、使用和配置。

如果您希望在您的 Linkis 服务上使用 ElasticSearch 引擎连接器，您需要安装 ElasticSearch 服务并保证服务可用。

通过如下命令验证 ElasticSearch 引擎连接器服务是否可用，如服务已开启用户验证则需要增加 --user username:password

  1. curl [--user username:password] http://ip:port/_cluster/healty?pretty

输出如下内容代表 ElasticSearch 服务可用，注意集群 status 为 green

  1. {
  2.   "cluster_name" : "docker-cluster",
  3.   "status" : "green",
  4.   "timed_out" : false,
  5.   "number_of_nodes" : 1,
  6.   "number_of_data_nodes" : 1,
  7.   "active_primary_shards" : 7,
  8.   "active_shards" : 7,
  9.   "relocating_shards" : 0,
  10.   "initializing_shards" : 0,
  11.   "unassigned_shards" : 0,
  12.   "delayed_unassigned_shards" : 0,
  13.   "number_of_pending_tasks" : 0,
  14.   "number_of_in_flight_fetch" : 0,
  15.   "task_max_waiting_in_queue_millis" : 0,
  16.   "active_shards_percent_as_number" : 100.0
  17. }

方式一：直接下载引擎连接器包

Linkis 引擎连接器下载

方式二：单独编译引擎连接器（需要有 maven 环境）

  1. # 编译
  2. cd ${linkis_code_dir}/linkis-engineconn-plugins/elasticsearch/
  3. mvn clean install
  4. # 编译出来的引擎连接器包，位于如下目录中
  5. ${linkis_code_dir}/linkis-engineconn-plugins/elasticsearch/target/out/

EngineConnPlugin 引擎连接器安装

将 2.1 中的引擎连接器包上传到服务器的引擎连接器目录下

  1. ${LINKIS_HOME}/lib/linkis-engineconn-plugins

上传后目录结构如下所示

  1. linkis-engineconn-plugins/
  2. ├── elasticsearch
  3. │   ├── dist
  4. │   │   └── 7.6.2
  5. │   │       ├── conf
  6. │   │       └── lib
  7. │   └── plugin
  8. │       └── 7.6.2

通过重启 linkis-cg-linkismanager 服务刷新引擎连接器

  1. cd ${LINKIS_HOME}/sbin
  2. sh linkis-daemon.sh restart cg-linkismanager

可以查看数据库中的 linkis_engine_conn_plugin_bml_resources 这张表的last_update_time 是否为触发刷新的时间。

  1. #登陆到linkis的数据库 
  2. select * from linkis_cg_engine_conn_plugin_bml_resources;

-codeType 参数说明

  • essql：通过 SQL 脚本的方式执行 ElasticSearch 引擎连接器任务
  • esjson：通过 JSON 脚本的方式执行 ElasticSearch 引擎连接器任务

essql 方式示例

注意： 使用这种形式， ElasticSearch 服务必须安装SQL插件，安装方式参考：https://github.com/NLPchina/elasticsearch-sql#elasticsearch-762

  1.  sh ./bin/linkis-cli -submitUser hadoop \
  2.  -engineType elasticsearch-7.6.2 -codeType essql \
  3.  -code '{"sql": "select * from kibana_sample_data_ecommerce limit 10' \
  4.  -runtimeMap linkis.es.http.method=GET \
  5.  -runtimeMap linkis.es.http.endpoint=/_sql \
  6.  -runtimeMap linkis.es.datasource=hadoop  \
  7.  -runtimeMap linkis.es.cluster=127.0.0.1:9200

esjson 方式示例

  1. sh ./bin/linkis-cli -submitUser hadoop \
  2. -engineType elasticsearch-7.6.2 -codeType esjson \
  3. -code '{"query": {"match": {"order_id": "584677"}}}' \
  4. -runtimeMap linkis.es.http.method=GET \
  5. -runtimeMap linkis.es.http.endpoint=/kibana_sample_data_ecommerce/_search \
  6. -runtimeMap linkis.es.datasource=hadoop  \
  7. -runtimeMap linkis.es.cluster=127.0.0.1:9200

更多 Linkis-Cli 命令参数参考： Linkis-Cli 使用

配置默认值是否必须说明
linkis.es.cluster127.0.0.1:9200是ElasticSearch 集群，多个节点使用逗号分隔
linkis.es.datasourcehadoop是ElasticSearch datasource
linkis.es.username无否ElasticSearch 集群用户名
linkis.es.password无否ElasticSearch 集群密码
linkis.es.auth.cachefalse否客户端是否缓存认证
linkis.es.sniffer.enablefalse否客户端是否开启 sniffer
linkis.es.http.methodGET否调用方式
linkis.es.http.endpoint/_search否JSON 脚本调用的 Endpoint
linkis.es.sql.endpoint/_sql否SQL 脚本调用的 Endpoint
linkis.es.sql.format{“query”:”%s”}否SQL 脚本调用的模板，%s 替换成 SQL 作为请求体请求Es 集群
linkis.es.headers.*无否客户端 Headers 配置
linkis.engineconn.concurrent.limit100否引擎连接器最大并发

如果默认参数不满足时，有如下几中方式可以进行一些基础参数配置

ElasticSearch - 图1

注意: 修改 IDE 标签下的配置后需要指定 -creator IDE 才会生效（其它标签类似），如：

  1. sh ./bin/linkis-cli -creator IDE -submitUser hadoop \
  2. -engineType elasticsearch-7.6.2 -codeType esjson \
  3. -code '{"query": {"match": {"order_id": "584677"}}}' \
  4. -runtimeMap linkis.es.http.method=GET \
  5. -runtimeMap linkis.es.http.endpoint=/kibana_sample_data_ecommerce/_search

提交任务接口，通过参数 params.configuration.runtime 进行配置

  1. http 请求参数示例 
  2. {
  3.     "executionContent": {"code": "select * from kibana_sample_data_ecommerce limit 10;", "runType":  "essql"},
  4.     "params": {
  5.                     "variable": {},
  6.                     "configuration": {
  7.                             "runtime": {
  8.                                 "linkis.es.cluster":"http://127.0.0.1:9200",
  9.                                 "linkis.es.datasource":"hadoop",
  10.                                 "linkis.es.username":"",
  11.                                 "linkis.es.password":""
  12.                                 }
  13.                             }
  14.                     },
  15.     "labels": {
  16.         "engineType": "elasticsearch-7.6.2",
  17.         "userCreator": "hadoop-IDE"
  18.     }
  19. }

通过修改目录 ${LINKIS_HOME}/lib/linkis-engineconn-plugins/elasticsearch/dist/7.6.2/conf/ 中的 linkis-engineconn.properties 文件进行配置，如下图：

ElasticSearch - 图2

Linkis 是通过引擎连接器标签来进行管理的，所涉及的数据表信息如下所示。

  1. linkis_ps_configuration_config_key:  插入引擎连接器的配置参数的key和默认values
  2. linkis_cg_manager_label：插入引擎连接器label如：elasticsearch-7.6.2
  3. linkis_ps_configuration_category： 插入引擎连接器的目录关联关系
  4. linkis_ps_configuration_config_value： 插入引擎连接器需要展示的配置
  5. linkis_ps_configuration_key_engine_relation:配置项和引擎连接器的关联关系

表中与引擎连接器相关的初始数据如下

  1. -- set variable
  2. SET @ENGINE_LABEL="elasticsearch-7.6.2";
  3. SET @ENGINE_ALL=CONCAT('*-*,',@ENGINE_LABEL);
  4. SET @ENGINE_IDE=CONCAT('*-IDE,',@ENGINE_LABEL);
  5. SET @ENGINE_NAME="elasticsearch";
  7. -- engine label
  8. insert into `linkis_cg_manager_label` (`label_key`, `label_value`, `label_feature`, `label_value_size`, `update_time`, `create_time`) VALUES ('combined_userCreator_engineType', @ENGINE_ALL, 'OPTIONAL', 2, now(), now());
  9. insert into `linkis_cg_manager_label` (`label_key`, `label_value`, `label_feature`, `label_value_size`, `update_time`, `create_time`) VALUES ('combined_userCreator_engineType', @ENGINE_IDE, 'OPTIONAL', 2, now(), now());
  11. select @label_id := id from `linkis_cg_manager_label` where label_value = @ENGINE_IDE;
  12. insert into `linkis_ps_configuration_category` (`label_id`, `level`) VALUES (@label_id, 2);
  14. -- configuration key
  15. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`) VALUES ('linkis.es.cluster', '例如:http://127.0.0.1:9200', '连接地址', 'http://127.0.0.1:9200', 'None', '', @ENGINE_NAME, 0, 0, 1, '数据源配置');
  16. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`) VALUES ('linkis.es.datasource', '连接别名', '连接别名', 'hadoop', 'None', '', @ENGINE_NAME, 0, 0, 1, '数据源配置');
  17. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`) VALUES ('linkis.es.username', 'username', 'ES集群用户名', '无', 'None', '', @ENGINE_NAME, 0, 0, 1, '数据源配置');
  18. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`) VALUES ('linkis.es.password', 'password', 'ES集群密码', '无', 'None', '', @ENGINE_NAME, 0, 0, 1, '数据源配置');
  19. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`) VALUES ('linkis.es.auth.cache', '客户端是否缓存认证', '客户端是否缓存认证', 'false', 'None', '', @ENGINE_NAME, 0, 0, 1, '数据源配置');
  20. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`) VALUES ('linkis.es.sniffer.enable', '客户端是否开启 sniffer', '客户端是否开启 sniffer', 'false', 'None', '', @ENGINE_NAME, 0, 0, 1, '数据源配置');
  21. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`) VALUES ('linkis.es.http.method', '调用方式', 'HTTP请求方式', 'GET', 'None', '', @ENGINE_NAME, 0, 0, 1, '数据源配置');
  22. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`) VALUES ('linkis.es.http.endpoint', '/_search', 'JSON 脚本调用的 Endpoint', '/_search', 'None', '', @ENGINE_NAME, 0, 0, 1, '数据源配置');
  23. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`) VALUES ('linkis.es.sql.endpoint', '/_sql', 'SQL 脚本调用的 Endpoint', '/_sql', 'None', '', @ENGINE_NAME, 0, 0, 1, '数据源配置');
  24. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`) VALUES ('linkis.es.sql.format', 'SQL 脚本调用的模板，%s 替换成 SQL 作为请求体请求Es 集群', '请求体', '{"query":"%s"}', 'None', '', @ENGINE_NAME, 0, 0, 1, '数据源配置');
  25. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`) VALUES ('linkis.es.headers.*', '客户端 Headers 配置', '客户端 Headers 配置', '无', 'None', '', @ENGINE_NAME, 0, 0, 1, '数据源配置');
  26. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `engine_conn_type`, `is_hidden`, `is_advanced`, `level`, `treeName`) VALUES ('linkis.engineconn.concurrent.limit', '引擎连接器最大并发', '引擎连接器最大并发', '100', 'None', '', @ENGINE_NAME, 0, 0, 1, '数据源配置');
  28. -- key engine relation
  29. insert into `linkis_ps_configuration_key_engine_relation` (`config_key_id`, `engine_type_label_id`)
  30. (select config.id as config_key_id, label.id AS engine_type_label_id FROM `linkis_ps_configuration_config_key` config
  31. INNER JOIN `linkis_cg_manager_label` label ON config.engine_conn_type = @ENGINE_NAME and label_value = @ENGINE_ALL);
  33. -- engine default configuration
  34. insert into `linkis_ps_configuration_config_value` (`config_key_id`, `config_value`, `config_label_id`)
  35. (select relation.config_key_id AS config_key_id, '' AS config_value, relation.engine_type_label_id AS config_label_id FROM `linkis_ps_configuration_key_engine_relation` relation
  36. INNER JOIN `linkis_cg_manager_label` label ON relation.engine_type_label_id = label.id AND label.label_value = @ENGINE_ALL);