Apache Flume之正则过滤器-低调大师

Apache Flume之正则过滤器

2017-08-01 551

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

执行所有这些步骤的流行工具之一是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,alok,mumbai,manager
2,jatin,chennai,manager

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

3,yogesh,kolkata,developer
5,jyotsana,pune,developer

如何达到这个要求

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

## Describe regex_filter interceptor and configure exclude events attribute
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = regex_filter
a1.sources.r1.interceptors.i1.regex = developer
a1.sources.r1.interceptors.i1.excludeEvents = true

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

a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
## assumption is that Hadoop is CDH
a1.sinks.k1.hdfs.path = hdfs://quickstart.cloudera:8020/user/hive/warehouse/managers
a1.sinks.k1.hdfs.fileType= DataStream
a1.sinks.k1.hdfs.writeFormat = Text

如何运行示例

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

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

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

a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444

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

telnet localhost 40000

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

1,alok,mumbai,manager
2,jatin,chennai,manager
3,yogesh,kolkata,developer
4,ragini,delhi,manager
5,jyotsana,pune,developer
6,valmiki,banglore,manager

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

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

# Name the components on this agent
a1.sources = r1
a1.sinks = k1
a1.channels = c1
# Describe/configure the source - netcat
a1.sources.r1.type = netcat
a1.sources.r1.bind = localhost
a1.sources.r1.port = 44444
# Describe the HDFS sink
a1.channels = c1
a1.sinks = k1
a1.sinks.k1.type = hdfs
a1.sinks.k1.channel = c1
a1.sinks.k1.hdfs.path = hdfs://quickstart.cloudera:8020/user/hive/warehouse/managers
a1.sinks.k1.hdfs.fileType= DataStream
a1.sinks.k1.hdfs.writeFormat = Text
## Describe regex_filter interceptor and configure exclude events attribute
a1.sources.r1.interceptors = i1
a1.sources.r1.interceptors.i1.type = regex_filter
a1.sources.r1.interceptors.i1.regex = developer
a1.sources.r1.interceptors.i1.excludeEvents = true
# Use a channel which buffers events in memory
a1.channels.c1.type = memory
a1.channels.c1.capacity = 1000
a1.channels.c1.transactionCapacity = 100
# Bind the source and sink to the channel
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1

本文作者：佚名

来源：51CTO

微信关注我们

原文链接：https://yq.aliyun.com/articles/179158

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

实时离线融合在唯品会的进展：在实时技术、数据、业务中寻找平衡

实时大数据分析是指对规模巨大的数据进行分析，利用大数据技术高效的快速完成分析，达到近似实时的效果，更及时的反映数据的价值和意义。所有人都能理解数据的时效性对于数据的价值至关重要。以唯品会为例，唯品会已经有一整套非常成熟的离线数据仓库系统。这套系统对于业务有非常大的指导意义，但目前碰到的问题是如何将各种计算、报表加速，从原来天级别、小时级别，加速到近实时来。这是我们开始实时离线融合这个项目的缘由。该工作我们是从 2016 年下半年开始的，到目前为止它仍然只是一个半成品，因此这里面包含的很多内容并不是最终的结论，在多数情况下，它仅仅是以唯品会的特点为基础，而不一定能无缝地适用于其他公司产品。我们希望抛砖引玉，对大家有所俾益。 1. 时效性与大数据第一个问题是：什么是实时（real-time）? 什么是离线（offline）？很多时候，我

2017-08-02

662

Spark： Apache Spark 是专为大规模数据处理而设计的快速通用的计算引擎。 Spark是UC Berkeley AMP lab (加州大学伯克利分校的AMP实验室)所开源的类Hadoop MapReduce的通用并行框架，Spark拥有Hadoop MapReduce所具有的优点；但不同于MapReduce的是Job中间输出结果可以保存在内存中，从而不再需要读写HDFS，因此Spark能更好地适用于数据挖掘与机器学习等需要迭代的MapReduce的算法。 Spark 是一种与 Hadoop 相似的开源集群计算环境，但是两者之间还存在一些不同之处，这些有用的不同之处使 Spark 在某些工作负载方面表现得更加优越，换句话说，Spark 启用了内存分布数据集，除了能够提供交互式查询外，它还可以优化迭代工作负载。 Spark 是在 Scala 语言中实现的，它将 Scala 用作其应用程序框架。与 Hadoop 不同，Spark 和 Scala 能够紧密集成，其中的 Scala 可以像操作本地集合对象一样轻松地操作分布式数据集。尽管创建 Spark 是为了支持分布式数据集上的...

2017-08-02

527

资源下载

更多资源

优质分享App

近一个月的开发和优化，本站点的第一个app全新上线。该app采用极致压缩，本体才4.36MB。系统里面做了大量数据访问、缓存优化。方便用户在手机上查看文章。后续会推出HarmonyOS的适配版本。

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。