当前位置： 首页 > news >正文

简述一下网站建设流程,域名备案网站购买,网站建设培训学院,软件开发培训机构去哪个学校深圳NiFi简介 Apache NiFi 是一款强大的开源数据集成工具#xff0c;旨在简化数据流的管理、传输和自动化。它提供了直观的用户界面和可视化工具#xff0c;使用户能够轻松设计、控制和监控复杂的数据流程#xff0c;NiFi 具备强大的扩展性和可靠性#xff0c;可用于处理海量数…NiFi简介 Apache NiFi 是一款强大的开源数据集成工具旨在简化数据流的管理、传输和自动化。它提供了直观的用户界面和可视化工具使用户能够轻松设计、控制和监控复杂的数据流程NiFi 具备强大的扩展性和可靠性可用于处理海量数据并且能很好地应对复杂的数据转换需求还可以设置定时调度任务 特点 可视化操作提供了图形化界面用户可以通过拖放组件来构建数据处理流程。数据处理功能强大能够实现数据的获取、转换、分发等操作。例如可以从各种数据源获取数据对数据进行格式转换、内容过滤等处理再将数据发送到目标系统。可靠性高在数据传输和处理过程中具有良好的容错机制保障数据的完整性和准确性。扩展性好可以轻松地扩展来处理大规模的数据流量和复杂的数据处理任务 NiFi的下载和安装 在本篇文章中主要讲述了四种部署方式一种是单机部署比较简单第二种是使用nifi自带的zookeeper部署伪分布模式,如果你的资源不足的话可以选用该模式第三种是使用nifi自带的zookeeper部署集群模式当然这种使用内部自带zookeeper的情况在实际开发中不常用第三种就是比较常用的使用外部zookeeper的nifi集群模式的部署 首先我们先来看一下单机部署 单机部署 以下部署要在java环境下若没有请事先安装jdk 下载安装包 进入Apache NiFi 官方网站来下载你所需要的版本 也可以直接点击nifi官网所有版本来下载 这里我下载的是1.13.2版本 下载完成后上传至Linux进行解压处理 解压tar -zxvf nifi-1.13.2-bin.tar.gz -C /opt/installs/ 修改配置文件 cp nifi-1.13.2/conf/nifi.properties nifi-1.13.2/conf/nifi.properties.bak 修改ip和端口号 启动nifi nifi-1.13.2/bin/nifi.sh 下面可以看到有启动/停止/运行/重启/状态等服务 启动nifinifi-1.13.2/bin/nifi.sh start 进入nifi的可视化界面 http://192.168.233.128:5800/nifi/ 日志的位置 Nifi 集群部署-伪分布模式 由于nifi内置zookeeper故我们先使用内置zookeeper进行搭建可以在一台服务器上搭建伪分布模式也可以在三台服务器上搭建集群模式因为伪分布模式在一台机器上搭建所以不同节点的相同功能端口会不同如果搭建集群模式IP不同那么不同节点的相同功能端口可以相同此处搭建集群模式 1、修改linux的⼀个安全机制 1进入vi /etc/selinux/config添加SELINUXdisabled防止后续出现一些问题 三台都需要修改改完重启一下 2因为配置完单机模式后nifi会产生许多新的数据库来存储数据为了防止对伪分布模式有影响这里先删除单机模式的nifi重新解压一份 解压tar -zxvf nifi-1.13.2-bin.tar.gz -C /opt/installs/重命名mv nifi-1.13.2/ nifi rm -rf /opt/installs/nifi-1.13.2/ 2、准备三个单机NIFI实例 nifi-1 nifi-2 nifi-3 3、修改配置文件 cp nifi/conf/nifi.properties nifi/conf/nifi.properties.bak 1修改三台服务器nifi中的zookeeper.properties

1节点21812节点21821节点2183clientPort12181 # 不同机器使用不同IPserver.1bigdata01:12888:13888server.2bigdata01:14888:15888server.3bigdata01:16888:17888

2新建nifi-1/state/zookeepernifi-2/state/zookeepernifi-3/state/zookeeper在此文件夹中新建文件myid分别对应写入1、2、3 3编辑节点conf/nifi.properties文件 1 ####################2 # State Management # 3 ####################4 nifi.state.management.configuration.file./conf/state-management.xml 5 nifi.state.management.provider.locallocal-provider 6 nifi.state.management.provider.clusterzk-provider7 # 指定此NiFi实例是否应运行嵌入式ZooKeeper服务器默认是false 8 nifi.state.management.embedded.zookeeper.starttrue 9 nifi.state.management.embedded.zookeeper.properties./conf/zookeeper.properties 10 11 # web properties #
12 nifi.web.war.directory./lib
13 # HTTP主机。默认为空白
14 nifi.web.http.hostbigdata01 15 # HTTP端口。默认值为8080 16 nifi.web.http.port18001 17 18 # cluster node properties (only configure for cluster nodes) #
19 # 如果实例是群集中的节点请将此设置为true。默认值为false 20 nifi.cluster.is.nodetrue 21 # 节点的完全限定地址。默认为空白 22 nifi.cluster.node.addressbigdata01 23 # 节点的协议端口。默认为空白 24 nifi.cluster.node.protocol.port28001 25 26 # 指定在选择Flow作为“正确”流之前等待的时间量。如果已投票的节点数等于nifi.cluster.flow.election.max.candidates属性指定的数量则群集将不会等待这么长时间。默认值为5 mins 27 nifi.cluster.flow.election.max.wait.time1 mins 28 # 指定群集中所需的节点数以便提前选择流。这允许群集中的节点避免在开始处理之前等待很长时间如果我们至少达到群集中的此数量的节点 29 nifi.cluster.flow.election.max.candidates1 30 31 # cluster load balancing properties #
32 nifi.cluster.load.balance.host 33 nifi.cluster.load.balance.port6342 34 35 # zookeeper properties, used for cluster management # 36 # 连接到Apache ZooKeeper所需的连接字符串。这是一个以逗号分隔的hostnameport对列表 37 nifi.zookeeper.connect.stringbigdata01:12181,bigdata01:12182,bigdata01:12183 38 nifi.zookeeper.connect.timeout3 secs
39 nifi.zookeeper.session.timeout3 secs
40 nifi.zookeeper.root.node/nifi bigdata01 143行nifi.web.http.hostbigdata01 144行nifi.web.http.port18001 241行nifi.cluster.is.nodetrue 242行nifi.cluster.node.addressbigdata01 243行nifi.cluster.node.protocol.port28001 256行nifi.cluster.load.balance.port6342 262行nifi.zookeeper.connect.stringbigdata01:12181,bigdata01:12182,bigdata01:12183 节点2,节点3内容跟节点1相同只是nifi.web.http.portnifi.cluster.node.protocol.portnifi.cluster.load.balance.port这三个端口区分开来避免端口重复 4修改conf/state-management.xml文件 61行property nameConnect Stringbigdata01:12181,bigdata01:12182,bigdata01:12183/property 4、启动三个实例启动完成后进入可视化界面 启动nifi-1/bin/nifi.sh start 启动nifi-2/bin/nifi.sh start 启动nifi-3/bin/nifi.sh start 可视化界面bigdata01:18001 Nifi 集群部署-内置Zookeeper 因为配置完单机模式后nifi会产生许多新的数据库来存储数据为了防止对集群模式有影响这里先删除单机模式的nifi重新解压一份重命名为nifi 1、分发nifi至三台服务器 xsync nifi/ 2、修改配置文件 1修改state-management.xml cluster-provideridzk-provider/idclassorg.apache.nifi.controller.state.providers.zookeeper.ZooKeeperStateProvider/classproperty nameConnect Stringbigdata01:2181,bigdata02:2181,bigdata03:2181/propertyproperty nameRoot Node/nifi/propertyproperty nameSession Timeout10 seconds/propertyproperty nameAccess ControlOpen/property/cluster-provider 进行分发 xsync /opt/installs/nifi/conf/state-management.xml 2修改nifi.properties bigdata01 51行nifi.state.management.embedded.zookeeper.starttrue默认为false如果使用外部的zookeeper集群为false 143行nifi.web.http.hostbigdata01 144行nifi.web.http.port18001 241行nifi.cluster.is.nodetrue 143行nifi.cluster.node.addressbigdata01 243行nifi.cluster.node.protocol.port28001 262行nifi.zookeeper.connect.stringbigdata01:2181,bigdata02:2181,bigdata03:2181 进行分发并修改主机名 xsync /opt/installs/nifi/conf/nifi.properties 修改143行、143行主机名为相应的另外两台服务器的名字即可 3修改zookeeper.properties clientPort2181
server.1bigdata01:2888:3888;2181 server.2bigdata02:2888:3888;2181 server.3bigdata03:2888:3888;2181 进行分发 xsync /opt/installs/nifi/conf/nifi.properties 4新建nifi/state/zookeeper在此文件夹中新建文件myid且在三台虚拟机中依次写入1、2、3 3、启动nifi并产看web界面 cd /opt/installs/nifibin/nifi.sh start #三台都要启动bigdata01:18001 Nifi 集群部署-外部Zookeeper 修改使用内置zookeeper状态下的nifi.properties中的第51行即可 web界面简介以及简单读取文件案例 选择自己想使用的处理器 填写文件夹路径三台服务器都要有 打勾了说明如果成功了就会停下来 把上述success打的勾取消再拖拽入一个output类型的处理器进行连接点击运行发现getFile运行中nifi的强大之处只要有一个处理器能运行就会执行此时我们点击运行并往上述文件夹中发送数据数据会先存储到管道中数据会管道把数据存到本地磁盘如果后续处理器发生错误也不会影响前面处理器的运行 点击小眼睛或者感叹号都可以查看内容 接下来配置输出路径会发现数据传到了输出路径中管道变成了0字节 从本地文件读取数据到MySQL 1、添加处理组file_to_mysql 开启hdfs服务start-dfs.sh 2、填写hadoop中core-site-xml和hdfs-site.xml的路径 /opt/installs/hadoop/etc/hadoop/core-site.xml,/opt/installs/hadoop/etc/hadoop/hdfs-site.xml 其他参数可以先不配置 因为默认会加载你hadoop的配置文件 点击运行往输入路径中写入文件 cp wcc.txt /home/data 离线同步MySQL数据到hdfs 类似于datax的功能 新建一个处理组mysql_to_hdfs 添加mysql端 添加驱动以及驱动地址 jdbc:mysql://bigdata01:3306/mydb01 com.mysql.jdbc.Driver /opt/installs/hive/lib/mysql-connector-java-8.0.26.jar 这是因为没有填写账号密码 再次启动即可 这时发现文件传输速度特别快可以修改调度时间因为我们的离线数仓是一天调度一次可以改成86400秒 查看hdfs的数据发现是乱码的 这时我们可以添加一个中间处理器将avro格式转换为json格式 添加之后再次启动即可 可以看到我们每次运行都把数据放入了同一个文件夹这样是不行的我们之前用datax导数据时放入了不同的文件夹 修改动态目录 添加dt后仍然无法识别需要再添加一个处理器UpdateAttribute 给前三个处理器执行一下使数据在第三个管道内 再运行第四个处理器在hdfs上查看 但是还是要手动去写能不能调用一些函数呢 双击可查看获取当前时间的方法 这里我们选用\({now():format(yyyy)}并改成年月日的形式\){now():format(yyyy-MM-dd)} 这样每天都会创建新的文件夹但是打开文件夹后发现里面的文件名是用UUID命名的虽然使文件名不重复但是不便于观看 以时间戳和后缀作为文件名这样可以控制文件的滚动 再次运行可以看到文件名得到了修改 还可以修改成每个小时生成一个新文件 实时监控kafka数据到hdfs 启动kafka集群添加kafka消费者选择对应的版本 修改配置 复制一份puthdfs修改输出路径 和上面一样通过添加日期函数使一个小时生成一个文件夹这样可以解决小文件问题