Pipeline

Pipeline 主要是用来对文件进行导入和导出的。本文主要介绍在 Linkis 中， Pipeline 引擎插件的安装、使用和配置。

方式一：直接下载引擎插件包

Linkis 引擎插件下载

方式二：单独编译引擎插件（需要有 maven 环境）

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

EngineConnPlugin 引擎插件安装

将 1.1 中的引擎插件包上传到服务器的引擎目录下

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

上传后目录结构如下所示

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

通过重启 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;

因为 Pipeline 引擎主要用来导入导出文件为主，现在我们假设从A向B导入文件为介绍案例

  1. sh bin/linkis-cli -submitUser  hadoop  \
  2. -engineType pipeline-1  -codeType pipeline  \
  3. -code "from hdfs:///000/000/000/A.dolphin  to file:///000/000/000/B.csv"

from hdfs:///000/000/000/A.dolphin to file:///000/000/000/B.csv 该内容 2.3 有解释

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

配置默认值是否必须说明
pipeline.output.moldcsv否结果集导出类型
pipeline.field.split,否csv分隔符
pipeline.output.charsetgbk否结果集导出字符集
pipeline.output.isoverwritetrue否是否覆写
wds.linkis.rm.instance3否pipeline引擎最大并发数
pipeline.output.shuffle.null.typeNULL否空值替换
wds.linkis.engineconn.java.driver.memory2g否pipeline引擎初始化内存大小

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

Pipeline - 图1

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

  1. sh bin/linkis-cli -creator IDE \
  2. -submitUser hadoop \
  3. -engineType pipeline-1  \
  4. -codeType pipeline \
  5. -code "from hdfs:///000/000/000/A.dolphin to file:///000/000/000/B.csv"

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

  1. http 请求参数示例 
  2. {
  3.     "executionContent": {"code": "from hdfs:///000/000/000/A.dolphin to file:///000/000/000/B.csv", "runType":  "pipeline"},
  4.     "params": {
  5.                     "variable": {},
  6.                     "configuration": {
  7.                             "runtime": {
  8.                                 "pipeline.output.mold":"csv",
  9.                                 "pipeline.output.charset":"gbk"
  10.                                 }
  11.                             }
  12.                     },
  13.     "labels": {
  14.         "engineType": "pipeline-1",
  15.         "userCreator": "hadoop-IDE"
  16.     }
  17. }

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

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

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

  1. -- set variable
  2. SET @PIPELINE_LABEL="pipeline-1";
  3. SET @PIPELINE_ALL=CONCAT('*-*,',@PIPELINE_LABEL);
  4. SET @PIPELINE_IDE=CONCAT('*-IDE,',@PIPELINE_LABEL);
  6. -- engine label
  7. insert into `linkis_cg_manager_label` (`label_key`, `label_value`, `label_feature`, `label_value_size`, `update_time`, `create_time`) VALUES ('combined_userCreator_engineType', @PIPELINE_ALL, 'OPTIONAL', 2, now(), now());
  8. insert into `linkis_cg_manager_label` (`label_key`, `label_value`, `label_feature`, `label_value_size`, `update_time`, `create_time`) VALUES ('combined_userCreator_engineType', @PIPELINE_IDE, 'OPTIONAL', 2, now(), now());
  10. select @label_id := id from linkis_cg_manager_label where `label_value` = @PIPELINE_IDE;
  11. insert into linkis_ps_configuration_category (`label_id`, `level`) VALUES (@label_id, 2);
  13. -- configuration key
  14. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `is_hidden`, `is_advanced`, `level`, `treeName`, `engine_conn_type`) VALUES ('pipeline.output.mold', '取值范围：csv或excel', '结果集导出类型','csv', 'OFT', '[\"csv\",\"excel\"]', '0', '0', '1', 'pipeline引擎设置', 'pipeline');
  15. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `is_hidden`, `is_advanced`, `level`, `treeName`, `engine_conn_type`) VALUES ('pipeline.field.split', '取值范围：，或\\t', 'csv分隔符',',', 'OFT', '[\",\",\"\\\\t\"]', '0', '0', '1', 'pipeline引擎设置', 'pipeline');
  16. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `is_hidden`, `is_advanced`, `level`, `treeName`, `engine_conn_type`) VALUES ('pipeline.output.charset', '取值范围：utf-8或gbk', '结果集导出字符集','gbk', 'OFT', '[\"utf-8\",\"gbk\"]', '0', '0', '1', 'pipeline引擎设置', 'pipeline');
  17. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `is_hidden`, `is_advanced`, `level`, `treeName`, `engine_conn_type`) VALUES ('pipeline.output.isoverwrite', '取值范围：true或false', '是否覆写','true', 'OFT', '[\"true\",\"false\"]', '0', '0', '1', 'pipeline引擎设置', 'pipeline');
  18. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `is_hidden`, `is_advanced`, `level`, `treeName`, `engine_conn_type`) VALUES ('wds.linkis.rm.instance', '范围：1-3，单位：个', 'pipeline引擎最大并发数','3', 'NumInterval', '[1,3]', '0', '0', '1', 'pipeline引擎设置', 'pipeline');
  19. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `is_hidden`, `is_advanced`, `level`, `treeName`, `engine_conn_type`) VALUES ('wds.linkis.engineconn.java.driver.memory', '取值范围：1-10，单位：G', 'pipeline引擎初始化内存大小','2g', 'Regex', '^([1-9]|10)(G|g)$', '0', '0', '1', 'pipeline资源设置', 'pipeline');
  20. INSERT INTO `linkis_ps_configuration_config_key` (`key`, `description`, `name`, `default_value`, `validate_type`, `validate_range`, `is_hidden`, `is_advanced`, `level`, `treeName`, `engine_conn_type`) VALUES ('pipeline.output.shuffle.null.type', '取值范围：NULL或者BLANK', '空值替换','NULL', 'OFT', '[\"NULL\",\"BLANK\"]', '0', '0', '1', 'pipeline引擎设置', 'pipeline');
  22. -- key engine relation
  23. insert into `linkis_ps_configuration_key_engine_relation` (`config_key_id`, `engine_type_label_id`)
  24. (select config.id as `config_key_id`, label.id AS `engine_type_label_id` FROM linkis_ps_configuration_config_key config
  25. INNER JOIN linkis_cg_manager_label label ON config.engine_conn_type = 'pipeline' and label_value = @PIPELINE_ALL);
  27. -- engine default configuration
  28. insert into `linkis_ps_configuration_config_value` (`config_key_id`, `config_value`, `config_label_id`)
  29. (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
  30. INNER JOIN linkis_cg_manager_label label ON relation.engine_type_label_id = label.id AND label.label_value = @PIPELINE_ALL);