流程图

kettle 面板图片

操作步骤

1、订阅数据源（kafka consumer）

2、抽取字段并转换key（JSON input）

3、判断img字段是否有值，有的话进行url转base64（JavaScript 代码）

// 获取输入字段的值
var imgUrl = pic1Url;var imgBase64 = '';
var connection = null; // 声明连接对象//获取当前时间
var createTime = new Date();// 检查 imgUrl 字段是否为空
if (imgUrl === null || imgUrl === undefined || imgUrl.trim() === "") {// 如果 imgUrl 字段为空，则添加常量字段imgBase64 = "";
} else {// 如果 imgUrl 字段不为空，则将 URL 转换为 Base64try {var url = new java.net.URL(imgUrl);connection = url.openConnection(); // 打开连接var inputStream = connection.getInputStream();var outputStream = new java.io.ByteArrayOutputStream();var buffer = java.lang.reflect.Array.newInstance(java.lang.Byte.TYPE, 1024); // 缓冲区大小var bytesRead;while ((bytesRead = inputStream.read(buffer)) != -1) {outputStream.write(buffer, 0, bytesRead); // 将数据写入输出流}imgBase64 = org.apache.commons.codec.binary.Base64.encodeBase64String(outputStream.toByteArray());inputStream.close(); // 关闭输入流} catch (e) {// 处理异常if (connection !== null) {connection.disconnect(); // 关闭连接}throw e; // 重新抛出异常} finally {if (connection !== null) {connection.disconnect(); // 关闭连接}}
}//生成id
var uuid = java.util.UUID.randomUUID().toString();var uniqueId = Date.now() + '-' + Math.random().toString(36).substring(2, 15);uniqueId;
// 将 imgBase64 设置到输出字段
imgBase64;
createTime;

4、添加部门code （add constants）

5、记录数据库（表输出）

6、将sysOrgCode跟base64加进去json，重新构造（JSON output、JSON input）

JSON output

‘

JSON input

7、发送到目标kafka （Kafka producer）