Apache Flume之正则过滤器

简介:

Apache Flume之正则过滤器

在当今的大数据世界中,应用程序产生大量的电子数据 – 这些巨大的电子数据存储库包含了有价值的、宝贵的信息。 对于人类分析师或领域专家,很难做出有趣的发现或寻找可以帮助决策过程的模式。 我们需要自动化的流程来有效地利用庞大的,信息丰富的数据进行规划和投资决策。 在处理数据之前,收集数据,聚合和转换数据是绝对必要的,并最终将数据移动到那些使用不同分析和数据挖掘工具的存储库中。

执行所有这些步骤的流行工具之一是Apache Flume。 这些数据通常是以事件或日志的形式存储。 Apache Flume有三个主要组件:

  • Source:数据源可以是企业服务器,文件系统,云端,数据存储库等。
  • Sink:Sink是可以存储数据的目标存储库。 它可以是一个集中的地方,如HDFS,像Apache Spark这样的处理引擎,或像ElasticSearch这样的数据存储库/搜索引擎。
  • Channel:在事件被sink消耗前由Channel 存储。 Channel 是被动存储。 Channel 支持故障恢复和高可靠性; Channel 示例是由本地文件系统和基于内存的Channel 支持的文件通道。

Flume是高度可配置的,并且支持许多源,channel,serializer和sink。它还支持数据流。 Flume的强大功能是拦截器,支持在运行中修改/删除事件的功能。支持的拦截器之一是regex_filter。

regex_filter将事件体解释为文本,并将其与提供的正则表达式进行对比,并基于匹配的模式和表达式,包括或排除事件。我们将详细看看regex_filter。

要求

从数据源中,我们以街道号,名称,城市和角色的形式获取数据。现在,数据源可能是实时流数据,也可能是任何其他来源。在本示例中,我已经使用Netcat服务作为侦听给定端口的源,并将每行文本转换为事件。要求以文本格式将数据保存到HDFS中。在将数据保存到HDFS之前,必须根据角色对数据进行过滤。只有经理的记录需要存储在HDFS中;其他角色的数据必须被忽略。例如,允许以下数据:


  
  
  1. 1,alok,mumbai,manager 
  2.  
  3. 2,jatin,chennai,manager  

下列的数据是不被允许的:


  
  
  1. 3,yogesh,kolkata,developer 
  2.  
  3. 5,jyotsana,pune,developer  

如何达到这个要求

可以通过使用 regex_filter 拦截器来实现。这个拦截器将根据规则基础来进行事件过滤,只有感兴趣的事件才会发送到对应的槽中,同时忽略其他的事件。


  
  
  1. ## Describe regex_filter interceptor and configure exclude events attribute 
  2.  
  3. a1.sources.r1.interceptors = i1 
  4.  
  5. a1.sources.r1.interceptors.i1.type = regex_filter 
  6.  
  7. a1.sources.r1.interceptors.i1.regex = developer 
  8.  
  9. a1.sources.r1.interceptors.i1.excludeEvents = true  

HDFS 槽允许数据存储在 HDFS 中,使用文本/序列格式。也可以使用压缩格式存储。


  
  
  1. a1.channels = c1 
  2.  
  3. a1.sinks = k1 
  4.  
  5. a1.sinks.k1.type = hdfs 
  6.  
  7. a1.sinks.k1.channel = c1 
  8.  
  9. ## assumption is that Hadoop is CDH 
  10.  
  11. a1.sinks.k1.hdfs.path = hdfs://quickstart.cloudera:8020/user/hive/warehouse/managers 
  12.  
  13. a1.sinks.k1.hdfs.fileType= DataStream 
  14.  
  15. a1.sinks.k1.hdfs.writeFormat = Text  

如何运行示例

首先,你需要 Hadoop 来让示例作为 HDFS 的槽来运行。如果你没有一个 Hadoop 集群,可以将槽改为日志,然后只需要启动 Flume。 在某个目录下存储 regex_filter_flume_conf.conf 文件然后使用如下命令运行代理。


  
  
  1. flume-ng agent --conf conf --conf-file regex_filter_flume_conf.conf --name a1 -Dflume.root.logger=INFO,console 

注意代理名称是 a1。我用了 Netcat 这个源。


  
  
  1. a1.sources.r1.type = netcat 
  2.  
  3. a1.sources.r1.bind = localhost 
  4.  
  5. a1.sources.r1.port = 44444  

一旦 Flume 代理启动,运行下面命令用来发送事件给 Flume。


  
  
  1. telnet localhost 40000 

现在我们只需要提供如下输入文本:


  
  
  1. 1,alok,mumbai,manager 
  2.  
  3. 2,jatin,chennai,manager 
  4.  
  5. 3,yogesh,kolkata,developer 
  6.  
  7. 4,ragini,delhi,manager 
  8.  
  9. 5,jyotsana,pune,developer 
  10.  
  11. 6,valmiki,banglore,manager  

访问 HDFS 你会观察到 HDFS 在 hdfs://quickstart.cloudera:8020/user/hive/warehouse/managers 下创建了一个文件,文件只包含经理的数据。

完整的 flume 配置 — regex_filter_flume_conf.conf — 如下:


  
  
  1. Name the components on this agent 
  2.  
  3. a1.sources = r1 
  4.  
  5. a1.sinks = k1 
  6.  
  7. a1.channels = c1 
  8.  
  9. # Describe/configure the source - netcat 
  10.  
  11. a1.sources.r1.type = netcat 
  12.  
  13. a1.sources.r1.bind = localhost 
  14.  
  15. a1.sources.r1.port = 44444 
  16.  
  17. # Describe the HDFS sink 
  18.  
  19. a1.channels = c1 
  20.  
  21. a1.sinks = k1 
  22.  
  23. a1.sinks.k1.type = hdfs 
  24.  
  25. a1.sinks.k1.channel = c1 
  26.  
  27. a1.sinks.k1.hdfs.path = hdfs://quickstart.cloudera:8020/user/hive/warehouse/managers 
  28.  
  29. a1.sinks.k1.hdfs.fileType= DataStream 
  30.  
  31. a1.sinks.k1.hdfs.writeFormat = Text 
  32.  
  33. ## Describe regex_filter interceptor and configure exclude events attribute 
  34.  
  35. a1.sources.r1.interceptors = i1 
  36.  
  37. a1.sources.r1.interceptors.i1.type = regex_filter 
  38.  
  39. a1.sources.r1.interceptors.i1.regex = developer 
  40.  
  41. a1.sources.r1.interceptors.i1.excludeEvents = true 
  42.  
  43. # Use a channel which buffers events in memory 
  44.  
  45. a1.channels.c1.type = memory 
  46.  
  47. a1.channels.c1.capacity = 1000 
  48.  
  49. a1.channels.c1.transactionCapacity = 100 
  50.  
  51. # Bind the source and sink to the channel 
  52.  
  53. a1.sources.r1.channels = c1 
  54.  
  55. a1.sinks.k1.channel = c1  

完整的项目代码请看 这里。  


本文作者:佚名

来源:51CTO

相关文章
|
7月前
|
Shell
Flume【问题记录 01】【at org.apache.flume.node.Application.main(Application.java:xxx) 类问题整理+其他类型问题总结】【避坑指南】
【2月更文挑战第17天】Flume【问题记录 01】【at org.apache.flume.node.Application.main(Application.java:xxx) 类问题整理+其他类型问题总结】【避坑指南】
348 2
|
7月前
|
Java Linux
Flume【环境搭建 01】CentOS Linux release 7.5 安装配置 apache-flume-1.9.0 并验证
【2月更文挑战第16天】Flume【环境搭建 01】CentOS Linux release 7.5 安装配置 apache-flume-1.9.0 并验证
136 0
|
7月前
|
数据采集 消息中间件 缓存
Apache Flume及快速安装
Apache Flume及快速安装
100 0
|
分布式计算 Hadoop
flume报错记录:java.net.ConnectException:拒绝连接; For more details see: http://wiki.apache.org/hadoop/ConnectionRefused
flume报错记录:java.net.ConnectException:拒绝连接; For more details see: http://wiki.apache.org/hadoop/ConnectionRefused
669 0
flume报错记录:java.net.ConnectException:拒绝连接; For more details see:  http://wiki.apache.org/hadoop/ConnectionRefused
|
存储 Java Linux
Apache Flume-- 自定义 sink(扩展)--数据写入本地|学习笔记
Apache Flume-- 自定义 sink(扩展)--数据写入本地
Apache Flume-- 自定义 sink(扩展)--数据写入本地|学习笔记
|
监控 Java 关系型数据库
Apache Flume-自定义 source(扩展)--功能测试实现|学习笔记
快速学习 Apache Flume-自定义 source(扩展)--功能测试实现
 Apache Flume-自定义 source(扩展)--功能测试实现|学习笔记
|
SQL 监控 关系型数据库
Apache Flume-自定义 source(扩展)|学习笔记
快速学习 Apache Flume-自定义 source(扩展),具体实现代码逻辑 Flume 提供了很多内置的 source、sink、channel。但是在某些场合下,它自带的组件可能不满足需求,为此 Flume 官方也提供了相关的接口,我们可以按照它的接口和规范进行开发,实现自己的需求。
Apache Flume-自定义 source(扩展)|学习笔记
|
缓存 监控 Java
Apache Flume-自定义拦截器-功能实现|学习笔记
快速学习 Apache Flume-自定义拦截器-功能实现
Apache Flume-自定义拦截器-功能实现|学习笔记
|
算法 Java 编译器
Apache Flume-自定义拦截器-代码逻辑梳理|学习笔记
快速学习 Apache Flume-自定义拦截器-代码逻辑梳理
Apache Flume-自定义拦截器-代码逻辑梳理|学习笔记
|
存储 监控 大数据
Apache Flume- 自定义拦截器-需求描述|学习笔记
快速学习 Apache Flume- 自定义拦截器-需求描述
 Apache Flume- 自定义拦截器-需求描述|学习笔记

推荐镜像

更多