Kettle 流程组件

目录

空操作（什么也不做）

这个组件可以用“人如其名”来形容，真的是什么也不做。一般作为流程的终点或者错误分支（跳过错误）来使用，可以把他理解为“垃圾箱”。

Switch / case 组件

组件说明

对于多种类型的值进行不同的选择路径。可以让数据流从一路到多路。

组件预览

switch/case

组件详解

  • swith 字段：作为条件判断使用的字段
  • 使用字符串包含比较：如果使用，判断 case 值时就是模糊匹配，类似于include、contains、like这种语法
  • case 值数据类型：case 值的数据类型，把他理解为 kttle 会根据数据类型选择对应的方法去比对 case 值
  • case 值：根据设置的 case 值配置不同的步骤（流程）
  • 默认目标步骤：swith case没有匹配到任何 case 值时后处理流程的默认步骤，类似于default语法

注意：

  • 一定得有个默认的路径
  • 先产生目标步骤，再进行路径连接

组件示例

本案例模拟一个根据id值进行分发的场景。

  1. 转换

switch转换

  1. SQL 脚本

编写一组测试用数据

CREATE TABLE if not exists `test` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `sex` varchar(4) DEFAULT NULL,
  `age` int(3) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
truncate table test ; 

insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (1, 'mars', '男', 30);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (2, 'lucy', '女', 18);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (3, 'lily', '女', 18);
  1. 运行结果
  • id=1

id1

  • id=2

id1

  • default

id1

过滤记录

组件说明

通过使用一个表达式从输入行中过滤数据，将结果是TURE或FALSE的行输出到 不同的节点。表达式是 "field" "OPERATOR" "field"/"value' 的形式，其中OPERATOR 可以是=，<>, <, >, <=, >=, REGEXP,IS NULL,IS NOT NULL, IN LIST, CONTAINS,STARTS WITH, ENDS WITH 。 用户可以增加多个表达式，并用AND或OR连接。

组件预览

过滤记录

组件详解

  • 发送true数据给步骤：当表达式的结果为true时执行的步骤
  • 发送false数据给步骤：当表达式的结果为false时执行的步骤
  • 条件：设置not时使用

注意： 发送true数据给步骤、发送false数据给步骤可以为空，即忽略这个步骤。同时为空时流程就中止了，相当于组件没有作用！！！

组件示例

本案例将模拟一个根据自定义条件过滤的分发条件的场景

  1. 转换

过滤转换

  1. 执行SQL脚本

准备一组测试数据

CREATE TABLE if not exists `test` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `sex` varchar(4) DEFAULT NULL,
  `age` int(3) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE if not exists `test1` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `sex` varchar(4) DEFAULT NULL,
  `age` int(3) DEFAULT NULL,
  `flagfield` varchar(20) default null,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

truncate table test ; 
truncate table test1 ;

insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (1, 'mars', '男', 30);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (2, 'lucy', '女', 18);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (3, 'lily', '女', 18);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (4, 'lilei', '女', 4);
  1. 运行结果
  • 表输出结果

表输出结果

  • 写日志结果

写日志结果

中止组件

组件说明

中止是数据流的终点，如果有数据到这里，将会报错。

组件预览

中止组件

组件详解

  • Options：中止时的操作
    • Abort the running transformation：中止正在运行的转换
    • Abort and log as an error：中止并在日志中记录为错误
    • Stop input processing：中止进程
  • 中止记录值：中止的条数，即有几条数据到中止组件后就中止

组件案例

本案例模拟一个id=1时中止流程

  1. 转换

中止组件转换

  1. 执行SQL脚本

编写一组测试用数据

CREATE TABLE if not exists `test` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `sex` varchar(4) DEFAULT NULL,
  `age` int(3) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
truncate table test ; 

insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (1, 'mars', '男', 30);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (2, 'lucy', '女', 18);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (3, 'lily', '女', 18);
  1. 运行结果
  • 转换日志

转换日志

  • 中止预览

中止预览

追加流

组件说明

把两个数据流按照指定的前后顺序合并到一起。

组件预览

流追加

组件详解

  • 步骤名：流程名称
  • 前：流追加后在前面的数据流
  • 后：流追加后在后面的数据流

组件示例

本案例模拟将2个数据流追加到一个数据流

  1. 转换

流追加转换

  1. 转换结果
  • 追加记录xls

追加记录xls

  • 获取test表数据 获取test表数据

  • 流追加

流追加

数据流优先级排序

组件说明

把N个数据流按照指定顺序合并到一起。流追加的plus版本~

组件预览

数据流有限级别排序

组件详解

  • 步骤名称：被排序的数据流名称

组件示例

  1. 转换

数据流优先级排序转换

  1. 执行SQL脚本

初始化测试数据

CREATE TABLE if not exists `test` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `sex` varchar(4) DEFAULT NULL,
  `age` int(3) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE if not exists `test1` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `sex` varchar(4) DEFAULT NULL,
  `age` int(3) DEFAULT NULL,
  `flagfield` varchar(20) default null,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE if not exists `test2` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `sex` varchar(4) DEFAULT NULL,
  `age` int(3) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

truncate table test ; 
truncate table test1 ;
truncate table test2 ;

insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (1, 'mars', '男', 30);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (2, 'lucy', '女', 18);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (3, 'lily', '女', 18);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (4, 'lilei', '女', 4);

insert INTO `demo`.`test1` (`id`, `name`, `sex`, `age`) VALUES (5, 'mars', '男', 30);
insert INTO `demo`.`test1` (`id`, `name`, `sex`, `age`) VALUES (6, 'lucy', '女', 18);

insert INTO `demo`.`test2` (`id`, `name`, `sex`, `age`) VALUES (7, 'lily', '女', 18);
insert INTO `demo`.`test2` (`id`, `name`, `sex`, `age`) VALUES (8, 'lilei', '女', 4);
  1. 转换结果

转换结果

根据Java代码过滤记录

组件说明

根据 java 表达式内容过滤记录，匹配上表达式的发送给“接收匹配的行的步骤”，否则发给“接收不匹配的行的步骤”。约等于过滤记录，只不过是用 Java 实现的条件~

组件预览

根据Java代码过滤记录

组件详解

  • 接收匹配的行的步骤：表达式返回true时执行的步骤
  • 接收不匹配的行的步骤：表达式返回false时执行的步骤
  • 条件 （Java 表达式）：Java 表达式，即 Java 编写的条件

组件示例

  1. 转换

根据Java代码过滤记录转换

  1. 执行SQL脚本

初始化测试数据

CREATE TABLE if not exists `test` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `sex` varchar(4) DEFAULT NULL,
  `age` int(3) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

CREATE TABLE if not exists `test1` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `sex` varchar(4) DEFAULT NULL,
  `age` int(3) DEFAULT NULL,
  `flagfield` varchar(20) default null,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

truncate table test ; 
truncate table test1 ;

insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (1, 'mars', '男', 30);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (2, 'lucy', '女', 18);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (3, 'lily', '女', 18);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (4, 'lilei', '女', 4);

insert INTO `demo`.`test1` (`id`, `name`, `sex`, `age`) VALUES (5, 'mars', '男', 30);
insert INTO `demo`.`test1` (`id`, `name`, `sex`, `age`) VALUES (6, 'lucy', '女', 18);
  1. 转换结果
  • true 记录

true记录

  • false 记录

false记录

检测空流

组件说明

检查数据流是否为空

组件预览

检测空留

组件详解

这个组件没啥可说的，就是检查连入的数据流是否为空

组件示例

  1. 转换

检测空流转换

  1. 执行SQL脚本

初始化测试数据

CREATE TABLE if not exists `test` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `sex` varchar(4) DEFAULT NULL,
  `age` int(3) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

truncate table test ; 

insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (1, 'mars', '男', 30);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (2, 'lucy', '女', 18);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (3, 'lily', '女', 18);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (4, 'lilei', '女', 4);
  1. 转换结果

检测空流转换

识别流的最后一行

组件说明

标记出来数据流里面最后一条数据。

组件预览

识别流的最后一行

组件详解

  • 字段结果：用来标记最后一行数据的字段

组件示例

  1. 转换

识别流的最后一行转换

  1. 执行SQL脚本

初始化测试数据

CREATE TABLE if not exists `test` (
  `id` int(11) NOT NULL,
  `name` varchar(255) DEFAULT NULL,
  `sex` varchar(4) DEFAULT NULL,
  `age` int(3) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

truncate table test ; 

insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (1, 'mars', '男', 30);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (2, 'lucy', '女', 18);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (3, 'lily', '女', 18);
insert INTO `demo`.`test` (`id`, `name`, `sex`, `age`) VALUES (4, 'lilei', '女', 4);
  1. 转换结果

识别流的最后一行转换结果

阻塞数据直到步骤都完成

组件说明

在转换中，因为转换的所有步骤都是同时启动的，所以用到阻塞这一个组件实现等待的需求。

组件预览

阻塞数据直到步骤都完成

组件详解

  • 转换步骤名称：步骤的名称，在单一转换中，名称必须唯一。

  • 监视下面的步骤：在此表格中指定要等待的步骤，可点击“获取步骤”按钮将转换中的全部步骤自动填充到表格中。注意： 需要先配置好“阻塞数据直到步骤都完成”后续步骤才可以。

  • 步骤名称：选择被监视的步骤名称。可多选，当选择的步骤全部执行完成后，阻塞步骤才通过。

  • 复制次数：步骤的副本号，当所选步骤具有多副本时（副本可通过右键点击步骤选择“改变开始复制的数量”进行设置，这个功能实际的含义是：设置多少份的拷贝，开多少线程。副本号从0开始），可设置副本号来指定等待某个特定的副本号执行完毕。如需等待所有副本完成，则需对每个副本设置一行，并使用此列对指定等待哪个副本步骤。

组件示例

  1. 转换

阻塞数据直到步骤都转换

  1. 转换结果

阻塞数据直到步骤都转换结果