Navigator发布审计数据到Kafka并使用Flume入HBase-CFANZ编程社区

温馨提示：如果使用电脑查看图片不清晰，可以使用手机打开文章单击文中的图片放大查看高清原图。

Fayson的github：

https://github.com/fayson/cdhproject

提示：代码块部分可以左右滑动查看噢

1.文档编写目的

Clodera Manager提供了Navigator Audit Server服务，该服务用于收集CDH集群中各个组件的审计日志（如：HDFS、HIVE、HBase、Impala、Solr等），需要查看各个组件的审计日志需要登录到Cloudera Navigator的管理界面去查看，审计日志的统计与分析方式也比较固定。企业用户需要将审计日志接入到自己的审计系统，这里Navigator Audit服务提供了两种方式一种是将审计事件推送至Kafka,另一种将审计事件推送至Syslog。本篇文章Fayson主要介绍配置Navigator Audit推送审计事件到Kafka并使用Flume入HBase。

Navigator发布审计数据到Kafka并使用Flume入HBase_cloudera

内容概述

1.环境准备

2.Navigator Audit配置修改

3.Flume配置采集Kafka数据

4.测试验证

测试环境说明

1.CM5.15.0和CDH5.14.2

2.CDK2.2.0

3.Flume NG1.6.0

4.集群已启用Kerberos

2.环境准备

1.集群已安装Kafka服务

Navigator发布审计数据到Kafka并使用Flume入HBase_cloudera_02

2.集群已安装Flume-NG服务

Navigator发布审计数据到Kafka并使用Flume入HBase_数据_03

3.在Cloudera Management Service中安装Navigator Metadata和Navigator Audit服务

Navigator发布审计数据到Kafka并使用Flume入HBase_cloudera_04

4.创建HBase的测试表

[root@cdh02 ~]# kinit hbase/admin
Password for hbase/admin@FAYSON.COM: 
[root@cdh02 ~]# hbase shell
hbase(main):001:0> create 'navigator_audits','info'

（可左右滑动）

Navigator发布审计数据到Kafka并使用Flume入HBase_cloudera_05

3.Navigator Audit配置Kafka

1.登录CM进入Cloudera Management Service的配置界面

Navigator发布审计数据到Kafka并使用Flume入HBase_kafka_06

2.搜索“Kafka”配置Navigator Audit与Kakfa集成

Navigator发布审计数据到Kafka并使用Flume入HBase_数据_07

Kafka的Topic名字可以自己指定，配置完成后重启Navigator Audit服务。

4.Flume Agent配置

由于集群启用了Kerberos，所以在配置时增加了Kerberos的配置信息，具体可以参考Fayson前面的文章《如何在Kerberos环境使用Flume采集Kafka数据并写入HDFS》，本篇文章中使用了Fayson自定义的HBaseSink在前面的文章《如何使用Flume采集Kafka数据写入HBase》有介绍使用，示例代码在GitHub：

https://github.com/fayson/cdhproject/tree/master/flumesink/src/main/java/com/cloudera/hbase

（可左右滑动）

1.配置Flume Agent读取Kafka数据并写入HBase

kafka.sources  = source1
kafka.channels = channel1
kafka.sinks = sink1
kafka.sources.source1.type = org.apache.flume.source.kafka.KafkaSource
kafka.sources.source1.kafka.bootstrap.servers = cdh02.fayson.com:9092,cdh03.fayson.com:9092,cdh04.fayson.com:9092
kafka.sources.source1.kafka.topics = NavigatorAuditEvents
kafka.sources.source1.kafka.consumer.group.id = flume-consumer
kafka.sources.source1.kafka.consumer.security.protocol= SASL_PLAINTEXT
kafka.sources.source1.kafka.consumer.sasl.mechanism= GSSAPI
kafka.sources.source1.kafka.consumer.sasl.kerberos.service.name= kafka

kafka.sources.source1.channels = channel1
kafka.channels.channel1.type = memory
kafka.channels.channel1.capacity = 10000
kafka.channels.channel1.transactionCapacity = 1000

kafka.sinks.sink1.channel = channel1
kafka.sinks.sink1.type = com.cloudera.hbase.FaysonHBaseSink
kafka.sinks.sink1.table = navigator_audits
kafka.sinks.sink1.columnFamily = info
#kafka.sinks.sink1.rowkeys = id,mobile_phone_num
kafka.sinks.sink1.serializer = com.cloudera.hbase.JsonHBaseEventSerializer
kafka.sinks.sink1.kerberosKeytab= /opt/cloudera/keytab/fayson.keytab
kafka.sinks.sink1.kerberosPrincipal= fayson@FAYSON.COM

（可左右滑动）

Navigator发布审计数据到Kafka并使用Flume入HBase_cloudera_08

2.配置Flume Agent启动加载jaas.conf

Navigator发布审计数据到Kafka并使用Flume入HBase_cloudera_09

jaas.conf配置文件内容如下：

[root@cdh02 ~]# more /opt/cloudera/keytab/jaas.conf 
KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   keyTab="/opt/cloudera/keytab/fayson.keytab"
   principal="fayson@FAYSON.COM";
};

Client {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   keyTab="/opt/cloudera/keytab/fayson.keytab"
   useTicketCache=false
   principal="fayson@FAYSON.COM";
};

（可左右滑动）

Navigator发布审计数据到Kafka并使用Flume入HBase_cloudera_10