分享

CDH5.1从Flume采集到索引(Solr)

desehawk 发表于 2015-8-25 11:01:44 [显示全部楼层] 回帖奖励 阅读模式 关闭右栏 2 26042



众所周知,CDH是Cloudera公司推出的基于稳定版的ApachHadoop环境。并且比官方的Hadoop更新更快。坚持季度update,年度release。其中集成的Hadoop生态系统中的所有组件互相兼容,并坚持更新其中的bug和feature,在学习测试、生产环境中都有很好的应用。
如果在生产环境中需要用到Hadoop,自行部署Hadoop,在其上再部署类似HBase、Flume、Impala、Hive等等,不但部署周期长,遇到问题在国内也很少能找到相应的解决办法。因为其相互的版本兼容就存在很大问题。所以笔者强烈推荐使用CDH的Hadoop。
CDH可以通过图形界面轻松部署Hadoop生态系统中的其他组件,对于初学者部署时间不超过两天。
本文旨在介绍在CDH5.1版本之上,使用Flume1.5采集文本数据,通过Morphline(ETL工具)进行分词,并存入Solr中创建索引。只通过配置即可,无需进行二次开发。

一、           环境搭建
1、 Linux:CentOS release6.5 X64(Final)
2、 CDH:CDH5,需要Cloudera Manager和CDH两个包,下载地址:http://archive-primary.cloudera.com/cm5/cm/5/
3、  JDK:1.7以上

二、           模拟应用场景
1、 从Flume的日志文件中摘取出部分内容形成一个原始的被采集文件。
2、 Flume采集源文件。
3、 通过编写正则表达式通过Flume自带的Morphline插件,将日志的每一行进行拆分。
4、 拆分成timestamp/classname/msg/message
5、 并且将timestamp再次拆分成
date_year/date_month/date_day/date_hour/date_minute/date_second
6、 将拆分结果存入Solr建立索引。
原始日志文件每行内容类似如下:


org.kitesdk.morphline.stdlib.LogInfoBuilder$LogInfo:found no grok match: [{message=[         atcom.typesafe.config.impl.Parseable.parse(Parseable.java:204)]}]


三、           搭建采集环境(Flume)
下图是Flume的工作原理结构图,简单了解一下。Flume-NG由三部分组成:Source、Channel、Sink。

Source负责采集源数据,将源数据放入Channel。
Channel负责暂存数据,供Sink使用。
Sink负责从Channel读取数据并根据需要发送出去。
以上三个部分Flume官方都提供多种类型可供使用,开发者可以根据不同需要选择不同类型的组件。并且都支持Plugin形式的二次开发,可以自定义Source、Channel和Sink。具体详细参见官方网站:http://flume.apache.org/FlumeUserGuide.html
进入Flumeà选择一个Slaveà配置
代理名称填写a5

[mw_shl_code=bash,true]a5.sources = r5//定义source
a5.sinks = k5//定义sink
a5.channels = c5//定义channel

a5.sources.r5.type = exec//定义source类型,为执行命令。
a5.sources.r5.channels = c5//指定source采集到的数据放入的channel
a5.sources.r5.command = tail -F/var/log/flume-ng/input_file_test.log//source采集的文件

a5.sources.r5.interceptors = i5_2 i5_3 i5_1//定义flume的拦截器
a5.sources.r5.interceptors.i5_3.type = static//拦截器类型为static(静态)
a5.sources.r5.interceptors.i5_3.key = BodyType//定义拦截器的key
a5.sources.r5.interceptors.i5_3.value = line//定义拦截器的value
a5.sources.r5.interceptors.i5_2.type = host//flume自带的拦截器(指定的被采集对象的ip地址)
a5.sources.r5.interceptors.i5_1.type = static
a5.sources.r5.interceptors.i5_1.key = sourceType
a5.sources.r5.interceptors.i5_1.value = fj

a5.sinks.k5.channel = c5//指定本sink从哪个channel中读取数据
a5.sinks.k5.type= org.apache.flume.sink.solr.morphline.MorphlineSolrSink//指定sink类型,MorphlineSolrSink
a5.sinks. k5.morphlineFile = morphlines.conf//指定morphline的配置文件路径,如果不加前面的路径,则默认读取该flume.conf文件所在同级路径的morphline文件。
a5.sinks. k5.morphlineId = morphline1
a5.sinks. k5.batchSize = 100//当channel中存在100条数据开始处理
a5.sinks. k5.batchDurationMillis = 1000//当channel中数据存在超过1000ms时开始处理。batchSize与batchDurationMillis采取优先原则,哪个参数值先到,哪个先处理。

a5.channels.c5.type = memory//指定channel类型,本例为内存
a5.channels.c5.capacity = 1000000//指定channel存储数据的能力,本例为1000000条。
a5.channels.c5.transactionCapacity = 100000//每个事务发给sink的最大event数。
a5.channels.c5.keep-alive = 30//指定channel存活时间30s,超过设定值将日出event。[/mw_shl_code]

四、           Morphline分词处理
笔者理解Morphline是一个ETL工具,在本例中的作用如下:
E(Extract):用readLine命令读取flume采集回的日志文件。
T(Transform):利用正则表达式对每行日志文件进行拆分。
L(Load):将拆分后的结果加载到Solr中创建索引。

Morphline配置文件说明详见代码中的注释。
[mw_shl_code=html,true]

SOLR_LOCATOR: {
         # Name of solr collection
         # 指定将索引创建到Solr中的哪个collection中。
         collection : collection1

         # ZooKeeper ensemble
         # 配置zookeeper地址
         zkHost :"slave2.CDH:2181,slave3.CDH:2181,slave4.CDH:2181/solr"
}

morphlines:
[
         {
                  #morphlineId,与Flume采集配置中的a5.sinks.k5.morphlineId对应
                  id : morphline1
                  # 导入commands的所有包,Morphline将所有的命令放在了org.kitesdk.包下
                  # 需要使用以下命令时必须导入这些命令所在的jar包。
                  # 与Solr命令相关的命令包在org.apache.solr下。
                  importCommands :["org.kitesdk.**", "org.apache.solr.**"]

                  commands : [
                  {
                          # Parse inputattachment and emit a record for each input line
                          # 读一行数据,并设置成UTF-8
                          readLine {
                                   charset :UTF-8
                          }
                  }
                  # 打印测试数据
                  { logInfo { format :"output record: {}", args : ["@{}"] } }
                  {
                  # 分词命令
                  grok {
                          # 导入字典数据,配置路径,可配置多个并用“,”分隔。
                          # grok-dictionary.conf中存储的是正则表达式采用K-V方式存储,
                          # 文件中定义原子正则表达式,新定义的正则可以引用其他的原子正则表达式。使用%{}方式引用。
                          # 读者可以详细参见该文件。
                          dictionaryFiles :[/home/flume-ng/conf/grok-dictionary.conf]
                          expressions : {
                                   # 拆分原始数据,以下为固定格式。flume会将读到的每行数据默认存入一个message的变量中。
                                   # 大括号中冒号前是grok-dictionary.conf文件中的正则标签,冒号后是solr中的field。
                                   message :"""%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:loglevel}%{DATA:classname}: %{GREEDYDATA:msg}"""
                          }
                  }

         }
         {
         # tryRules是异常处理功能,因为本例中要将时间戳做进一步拆分,拆分成年、月、日、时、分、秒等。
         # 不同数据的时间戳格式又不禁相同,所以可以对时间戳格式进行穷举。不同格式的时间戳就可以拆分了。
         tryRules {
                  throwExceptionIfAllRulesFailed: true
                  rules : [
                  {
                  commands : [
                  {
                          grok {
                                   dictionaryFiles: [/home/flume-ng/conf/grok-dictionary.conf]
                                   expressions :{
                                            timestamp:"%{YEAR:date_year}-%{MONTHNUM:date_month}-%{MONTHDAY:date_day}[T]%{HOUR:date_hour}:?%{MINUTE:date_minute}(?::?%{SECOND:date_second})"
                          }
                  }
                  }
                  ]
         }
         {
                  commands : [
                  {
                          grok {
                                   dictionaryFiles: [/home/flume-ng/conf/grok-dictionary.conf]
                                   expressions :{
                                   timestamp : "%{YEAR:date_year}-%{MONTHNUM:date_month}-%{MONTHDAY:date_day}[T]%{HOUR:date_hour}:?%{MINUTE:date_minute}(?::?%{SECOND:date_second})"
                  }
         }
         }
         ]
         }
         {
                  commands : [
                  {
                          grok {
                                   dictionaryFiles: [/home/flume-ng/conf/grok-dictionary.conf]
                                   expressions :{
                                            timestamp:"%{YEAR:date_year}/%{MONTHNUM:date_month}/%{MONTHDAY:date_day}[T]%{HOUR:date_hour}:?%{MINUTE:date_minute}(?::?%{SECOND:date_second})"
                          }
                  }
         }
         ]
         }
         ]
         }

}

         {
                  # 转换时间格式,将输入时间转换成Solr识别的格式"yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
                  convertTimestamp {
                          # Solr中的时间戳field:timestamp
                          field : timestamp
                          inputFormats :["yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", "yyyy-MM-ddHH:mm:ss","yyyy/MM/dd HH:mm:ss"]
                          outputFormat :"yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"
                  }
         }
         {
                  # Solr中的索引主键field:id
                  generateUUID {
                          field : id
                  }
         }

         # schema.xml.
         {
                  sanitizeUnknownSolrFields {
                  # Location from which to fetchSolr schema
                  solrLocator : ${SOLR_LOCATOR}

                  renameToPrefix :"ignored_"
                  }
         }

         # load the record into a SolrServer orMapReduce SolrOutputFormat.
         {
                  loadSolr {
                          solrLocator :${SOLR_LOCATOR}
                  }
         }
         ]
         }
]
[/mw_shl_code]
五、           编辑Solr的schema.xml文件
Solr的schema.xml文件一般存放在/etc/solr/instance.conf/conf目录。
编辑schema.xml文件
将必须的field进行更新。
[mw_shl_code=xml,true]<pre lang="html" line="1"escaped="true">
<field name="timestamp" type="date"indexed="true" stored="true"omitNorms="true"/>
<field name="date_year" type="int"indexed="true" stored="true"omitNorms="true"/>
<field name="date_month" type="int"indexed="true" stored="true"omitNorms="true"/>
<field name="date_day" type="int"indexed="true" stored="true"omitNorms="true"/>
<field name="date_hour" type="int"indexed="true" stored="true"omitNorms="true"/>
<field name="date_minute" type="int"indexed="true" stored="true"omitNorms="true"/>
<field name="date_second" type="string"indexed="true" stored="true"omitNorms="true"/>
<!—flume配置文件中官方提供的host拦截器-->
<field name="host" type="string"indexed="true" stored="true"omitNorms="true"/>
<!—flume配置文件中的一个静态拦截器-->
<field name="sourceType"type="string" indexed="true" stored="true" omitNorms="true"/>
<!—Morphline文件中的message-->
<field name="message" type="string"indexed="false" stored="true"omitNorms="true"/>
</pre>[/mw_shl_code]
以上文件编辑完成并保存后需要上传并重新加载一下这个collection。
使用如下命令:
[mw_shl_code=bash,true]solrctl instancedir --update myconfig/etc/solr/instance.conf
solrctl collection --reload collection1[/mw_shl_code]

以上即为在CDH环境中部署Flume采集数据,利用Morphline进行分词,并加载到Solr中创建索引的过程。
文中如有错误或更好的方案欢迎交流。
QQ:58431505  //认证请注明:Solr交流



已有(2)人评论

跳转到指定楼层
rainbow55 发表于 2018-7-17 11:01:42
您好,请问grok-dictionary.conf这个文件中的配置是什么,我在哪里可以找到呢?
回复

使用道具 举报

您需要登录后才可以回帖 登录 | 立即注册

本版积分规则

关闭

推荐上一条 /2 下一条