CDH-Kerberos环境下使用flume消费带Sentry认证的kafka数据保存到hdfs中

Posted 2021-04-14 大数据哔哔机

前提：

1.kerberos环境可用
2.flume服务可用
3.kafka服务可用并做了sentry权限管理
4.hdfs服务可用并集成了sentry权限管理

1 配置flume

打开CDH-Flume服务界面，点击配置，搜索配置。输入agent配置内容（注意代理名称要和配置文件中的一致！！！）

配置文件内容 ：

testkafka.channels = c1
testkafka.sources = s1
testkafka.sinks = k1

testkafka.sources.s1.type =org.apache.flume.source.kafka.KafkaSource
testkafka.sources.s1.kafka.bootstrap.servers=beta2.hadoop.com:9092,beta3.hadoop.com:9092,beta4.hadoop.com:9092
testkafka.sources.s1.kafka.topics = xt_test_flume2hdfs
testkafka.sources.s1.kafka.consumer.group.id = flume-consumer
testkafka.sources.s1.kafka.consumer.security.protocol= SASL_PLAINTEXT
testkafka.sources.s1.kafka.consumer.sasl.mechanism= GSSAPI
testkafka.sources.s1.kafka.consumer.sasl.kerberos.service.name= kafka
testkafka.sources.s1.channels = c1

testkafka.channels.c1.type = memory

testkafka.sinks.k1.type = hdfs
testkafka.sinks.k1.channel = c1
testkafka.sinks.k1.hdfs.kerberosKeytab= /flume-keytab/xiet.keytab
testkafka.sinks.k1.hdfs.kerberosPrincipal= [e-mail]
testkafka.sinks.k1.hdfs.path =/user/xiet/kafka-test20181102
testkafka.sinks.k1.hdfs.filePrefix = events-
testkafka.sinks.k1.hdfs.writeFormat = Text

配置文件中，指定了kafka的验证方式，为SASL_PLAINTEXT，需要配置Jaas.conf文件
写入HDFS需要kerberos认证，并且认证账户需要有写入路径的写入权限

在配置页面搜索：Flume Agent 的 Java 配置选项
配置agent的java配置，指定flume权限认证的配置，此处配置为/flume-keytab/flume_jaas.conf，要在相应的agent服务器创建此文件并配置。

到此在CM上的配置完毕，接下来到agent所在的服务器中配置flume的认证文件
/flume-keytab/flume_jaas.conf

KafkaClient {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  storeKey=true
  serviceName="kafka"
  keyTab="flume.keytab"
  principal="flume/[e-mail]";
};

2 kafka配置

创建角色flume
赋予相关topic权限给flume
此处的topic和consumer group和上一小节中的flume配置中保持一致！

[root@beta4 run-kafka]# kafka-sentry -cr -r flume
[root@beta4 run-kafka]# kafka-sentry -arg -r flume -g flume
[root@beta4 run-kafka]# kafka-sentry -gpr -r flume -p "Topic=xt_test_flume2hdfs->action=all"
[root@beta4 run-kafka]# kafka-sentry -gpr -r flume -p "CONSUMERGROUP=flume-consumer->action=all"

3 测试

完成上两小节的配置后，确保kafka和flume启动并为最新的配置项。
开始测试：
往topic中生产数据：

[root@beta4 run-kafka]# sh run.sh 
SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
0	xiet0	
1	xiet1	
2	xiet2	
3	xiet3	
4	xiet4	
5	xiet5	
6	xiet6	
7	xiet7	
8	xiet8	
9	xiet9	

在这里我发送了十条数据到kafka中，在cm查看flume的日志信息

显示成功，再到hue中查看hdfs目录

写入成功，测试完成！

--------------------- 

来源：CSDN 

原文：https://blog.csdn.net/weixin_35852328/article/details/83661279 

版权声明：本文为博主原创文章，转载请附上博文链接！