【大数据】NiFi 中的 Controller Service

news/2024/10/18 8:23:52/

NiFi 中的 Controller Service

  • 1.Service 简介
    • 1.1 Controller Service 的配置
      • 1.1.1 SETTING 基础属性
      • 1.1.2 PROPERTIES 使用属性
      • 1.1.3 COMMENT 页签
    • 1.2 Service 的使用范围
  • 2.全局参数配置
  • 3.DBCPConnectionPool 的使用样例
  • 4.在 ExcuseGroovyScript 组件中使用 Service

1.Service 简介

首先 NiFi 中的 Controller Service 和我们 MVC 概念中的 Controller Service 不是一个概念,NiFi 中的 Controller Service 更像是和 Processor 同级的一个概念,它和 Processor 在我个人的使用经验来理解的话就是 它是预制好的各种服务,可以被 Processor 引用或者支撑 Processor,例如一个 SQL 读取的 Processor,它得需要 JDBC 的连接,才能访问数据库。这里 Controller Service 就可以是一个 JDBC 的连接池服务。

同理,Controller Service 也是支持扩展的,可以像自定义开发 Processor 一样,根据自己的业务需求,进行自定义的 Controller Service 开发。

当我们使用某些依赖 Service 的组件(Processor)时,在配置中会出现选择 Service 或者创建新的 Service 的情况,这里的 Service 即是 NiFi 的 Controller Service,一旦创建新的,则会生成一个以 Group 为范围的 “全局” Service 对象,这时,再有依赖同类型 Service 的 Processor 时,可以直接选中。

在这里插入图片描述
在这里插入图片描述

1.1 Controller Service 的配置

单独查看 Controller Service 可以从面板空白处,右键 Configure 来看,如下图:

在这里插入图片描述
这是一个 JDBC 的连接池 Service,它包含的属性有 名称类型简介启用状态操作;从操作中可以看到配置该 Service 需要填写基本的各类属性;其中,Service 是有启停状态的,如果想修改 Service 的属性内容,必须先保证该 Service 是停用状态,然后点击配置标识,则进入配置页面,它的配置和 Processor 的差不多,通过页签区别,共有三个页签:SETTING基础属性)、PROPERTIES使用属性)、COMMENT页签)。

1.1.1 SETTING 基础属性

基础属性,包含左侧的名称,名称可以进行更改,右侧包含引用此 Service 的 Processor 列表。

1.1.2 PROPERTIES 使用属性

在这里插入图片描述
核心的业务配置,此标签页的配置项根据不同的 Service,配置内容不一致,具体的配置项以及使用,可以参考官方的文档;这里的是 JDBC 的连接池,所以基本需要连接数据库所需的 URL、数据库的账号密码、数据库的驱动类名称、驱动类的依赖 jar 包路径 ,这里不少 Service 可能都需要第三方的 jar 包依赖才可以使用,长期使用或生产环境下,建议将所有 jar 资源集中放在统一路径下。

1.1.3 COMMENT 页签

在这里插入图片描述
一个提供 Service 使用说明的页签,可根据自己实际需求,补充使用 Service 的用法以及描述。

1.2 Service 的使用范围

在 NiFi 中,Group 同时也对 Service 起作用,如果我们在一个 NiFi 的最外层的平面上新增 Controller Service,那么这些 Service 的作用域是整个 NiFi 的任何位置,如果我们在某个 Group 内创建 Controller Service, 那么这个 Controller Service 仅在 Group 范围内可以被引用,NiFi 的这种机制也是方便 Service 的使用和维护。

在这里插入图片描述

2.全局参数配置

类似于数据库连接池、Kafka、Redis 等各种组件的连接池、客户端 Client 的 Service 在实际的使用中会非常多,由此配置的 Service 也会非常多,于是就会产生很多次的反复配置 URL、账号这一系列重复的内容,由于 NiFi 的特性,这些 Service 又和组件(Processor)一样,四散在各处,这就使得维护和运维管理变得很繁琐,调试、调整、查看的时候,要不停的各个 Group 来回跳转、调整不同的 Service 的 Configure;为应对此类问题,NiFi 提供了全局配置的机制来弥补。

使用变量前:

在这里插入图片描述
这里的 URLDriver Class NameDatabase User 在实际生产环境中,可能都是固定的数据库和固定的服务,几乎不需要变的,可能只需要配置一遍就好,不需要每次创建 Service 都写一遍;所以可以这里可以使用上下文变量(Parameter Context)。

首先,打开 Parameter Context,创新一组新的变量:

在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
在这里插入图片描述
再进入 CONTROLLER SERVICES 对 Service 的配置进行修改,将具体的 URLDriver Class NameDatabase User 等参数,全部使用变量替换(变量使用 # 符 )。

在这里插入图片描述

3.DBCPConnectionPool 的使用样例

下面将使用 NiFi 实现一个简单的 Demo,从 MySQL 数据库中读取部分数据,将数据进行筛选,然后将数据输出;

首先,使用 ExecuteSQL 组件,读取 MySQL 中的数据,根据上文描述,创建一个 DBCPConnectionPool 的 Service,然后启动:

在这里插入图片描述
添加 ExecuteSQL 组件,配置相关内容,根据自定义编写的 SQL 读取数据库内容:

在这里插入图片描述
随后添加 ConvertAvroToJSON 组件,这里从数据库读出的数据是不可读的,为了方便查看调试、同时也是为了后续使用 Groovy 处理数据,所以选择转换为 JSON 进行处理,实际使用可以根据自身情况选择转换器:

在这里插入图片描述
添加 ExecuteGroovyScript 组件,使用 Groovy 脚本对数据进行处理。

在这里插入图片描述
Groovy 的脚本内容如下:

import org.apache.commons.io.IOUtils;
import java.nio.charset.StandardCharsets;
import groovy.json.JsonBuilder;
import groovy.json.JsonOutput;
import groovy.json.JsonSlurper;
import groovy.json.StringEscapeUtils;
import java.util.*;def dataJson = getInputJSONData()
if(null == dataJson){return;
}
def rss = []
for(int i = 0; i < dataJson.size();i++){def tem = dataJson.get(i);//在这里可以对数据进行处理rss.add(tem.name);
}// 输出
if(rss.size()>0){sendData(rss,REL_SUCCESS);
}/*** 读取输入流* @author GCC***/
def getInputJSONData(){def flow = session.get()if(null == flow){log.error("the flow is null ...");return;}def dataJson = null;def jsonStr = "";session.read(flow,{inputStream ->jsonStr = IOUtils.toString(inputStream, StandardCharsets.UTF_8)} as InputStreamCallback);try{dataJson = new JsonSlurper().parseText(jsonStr);}catch(Exception e){log.error("输入流格式错误")}session.remove(flow);return dataJson;
}/***输出数据至后续管道*@param result 输出的数据*@param outStream 输出的管道*@author GCC***/
void sendData(def result,def outStream){String successFlowFileStr = StringEscapeUtils.unescapeJavaScript(new JsonOutput().toJson(result).toString());def newflow = session.create();newflow = session.write(newflow, {outputStream ->outputStream.write(successFlowFileStr.getBytes(StandardCharsets.UTF_8))} as OutputStreamCallback)session.transfer(newflow, outStream);
}

最后使用 LogMessage 组件作为接收数据,实际情况可以将数据转为下一处理节点或存储等等。

在这里插入图片描述
在这里插入图片描述

4.在 ExcuseGroovyScript 组件中使用 Service

ExcuseGroovyScript 组件内部使用 Groovy 脚本处理数据时,可能需要再次读取数据库或者使用其他第三方数据来辅助处理,这时候,ExcuteGroovyScript 组件支持可以引入 Service,提供用户编写的 Groovy 脚本内部使用 Service;

首先需要在 ExcuteGroovyScript 组件的 PROPERTIES 配置中新增属性:

在这里插入图片描述
这里,添加属性时,会让用户输入用户给该属性的命名,如果是普通命名,这里的属性仅仅作为静态数据而已,但是如果使用关键字 SQL. 或者 CTL. 作为名称前缀时,则能够使用 Service,后续的属性值则会变成 Service 的选择。

在 Groovy 的代码中,则可以通过 SQL.mysql.{method} 的方式,调用 Service 的方法:

import org.apache.commons.io.IOUtils;
import java.nio.charset.StandardCharsets;
import groovy.json.JsonBuilder;
import groovy.json.JsonOutput;
import groovy.json.JsonSlurper;
import groovy.json.StringEscapeUtils;
import java.util.*;def dataJson = getInputJSONData()
if(null == dataJson){return;
}
def rss = []
for(int i = 0; i < dataJson.size();i++){def tem = dataJson.get(i);def mapdic = [:]// 使用 Service 查询数据库SQL.mysql.eachRow("SELECT id, value FROM tb_dic_detail WHERE u_status = 1 "){row->mapdic.put(row.id.toString(),row.value.toString());    }rss.add(tem.name);
}// 输出
if(rss.size()>0){sendData(rss,REL_SUCCESS);
}/***************************************************公共函数***************************************************//*** 读取输入流* @author GCC***/
def getInputJSONData(){def flow = session.get()if(null == flow){log.error("the flow is null ...");return;}def dataJson = null;def jsonStr = "";session.read(flow,{inputStream ->jsonStr = IOUtils.toString(inputStream, StandardCharsets.UTF_8)} as InputStreamCallback);try{dataJson = new JsonSlurper().parseText(jsonStr);}catch(Exception e){log.error("输入流格式错误")}session.remove(flow);return dataJson;
}/***输出数据至后续管道*@param result 输出的数据*@param outStream 输出的管道*@author GCC***/
void sendData(def result,def outStream){String successFlowFileStr = StringEscapeUtils.unescapeJavaScript(new JsonOutput().toJson(result).toString());def newflow = session.create();newflow = session.write(newflow, {outputStream ->outputStream.write(successFlowFileStr.getBytes(StandardCharsets.UTF_8))} as OutputStreamCallback)session.transfer(newflow, outStream);
}

http://www.ppmy.cn/news/1279099.html

相关文章

多级缓存:亿级流量的缓存方案

文章目录 一.多级缓存的引入二.JVM进程缓存三.Lua语法入门四.多级缓存1.OpenResty2.查询Tomcat3.Redis缓存预热4.查询Redis缓存5.Nginx本地缓存6.缓存同步 一.多级缓存的引入 传统缓存的问题 传统的缓存策略一般是请求到达Tomcat后&#xff0c;先查询Redis&#xff0c;如果未…

YOLO同时对图片和标签进行重命名

import osimage_folder /path/to/image/folder # 图像文件夹路径 label_folder /path/to/label/folder # 标签文件夹路径image_files sorted(os.listdir(image_folder)) # 获取图像文件列表并排序 label_files sorted(os.listdir(label_folder)) # 获取标签文件列表并排…

安装nodejs,配置环境变量并将npm设置淘宝镜像源

安装nodejs并将npm设置淘宝镜像源 1. 下载nodejs 个人不喜欢安装包&#xff0c;所以是下载zip包的方式。这里我下载的node 14解压包版本 下载地址如下&#xff1a;https://nodejs.org/dist/v14.15.1/node-v14.15.1-win-x64.zip 想要其他版本的小伙伴去https://nodejs.org/di…

Uniapp 开发 BLE

BLE 低功耗蓝牙&#xff08;Bluetooth Low Energy&#xff0c;或称Bluetooth LE、BLE&#xff0c;旧商标Bluetooth Smart&#xff09;&#xff0c;用于医疗保健、运动健身、安防、工业控制、家庭娱乐等领域。在如今的物联网时代下大放异彩&#xff0c;扮演者重要一环&#xff…

论文阅读——llava

Visual Instruction Tuning LLaVA 指令智能体分为两类&#xff1a;端到端的&#xff0c;通过LangChain[1]/LLM[35]协调各种模型的系统。 数据集生成用GPT辅助生成的&#xff0c;具体不写了。 模型结构&#xff1a; input image Xv LLM&#xff1a;Vicuna visual encoder&a…

【uniapp小程序-生成二维码+多个图片文字合并一张图】

<!-- 二维码 --><canvas id"qrcode" canvas-id"qrcode" width"120" ></canvas><!-- 生成带小程序码的分享图片 --><canvas canvas-id"shareCanvas" class"share-canvas"></canvas>#qrc…

计算机毕业设计------JSP教务处学生成绩管理系统

项目介绍 本项目包含管理员、教师、学生三种角色&#xff1b; 用户角色包含以下功能&#xff1a; 修改密码,查看自己的信息,查看自己的成绩,登录界面等功能。 管理员角色包含以下功能&#xff1a; 修改示例,增删改查学生信息,增删改查教师信息,增删改查课程信息,管理员修改…

大语言模型概述

在科技飞速发展的当今时代&#xff0c;人工智能技术成为社会进步的关键推动力之一。在广泛关注的人工智能领域中&#xff0c;大语言模型以其引人注目的特性备受瞩目。 大语言模型的定义及发展历史 大语言模型是一类基于深度学习技术的人工智能模型&#xff0c;旨在理解和生成自…