前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >如何在Kerberos环境使用Flume采集Kafka数据并写入HDFS

如何在Kerberos环境使用Flume采集Kafka数据并写入HDFS

作者头像
Fayson
修改2018-04-01 20:47:03
5.9K0
修改2018-04-01 20:47:03
举报
文章被收录于专栏:Hadoop实操

温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看。

注意:Fayson的github调整为:https://github.com/fayson/cdhproject,本文的代码在github中也能找到。

1.文档编写目的


在Kafka集群实际应用中,Kafka的消费者有很多种(如:应用程序、Flume、Spark Streaming、Storm等),本篇文章主要讲述如何在Kerberos环境使用Flume采集Kafka数据并写入HDFS。关于Flume更多sink方式实现敬请关注Fayson后续的文章。本文的数据流图如下:

  • 内容概述

1.Kafka集群启用Kerberos

2.环境准备及配置Flume Agent

3.java访问并测试

  • 测试环境

1.CM和CDH版本为5.11.2

2.采用root用户操作

  • 前置条件

1.集群已启用Kerberos

2.集群已安装Kafka

3.集群已安装Flume

2.Kafka集群启用Kerberos


登录Cloudera Manager进入Kafka服务,修改如下配置Kerberos.auth.enable和security.inter.broker.protocol配置为如下截图:

保存配置并重启Kafka服务。

3.环境准备


由于Kafka集群已启用Kerberos认证,这里需要准备访问Kafka集群的环境,如Keytab、jaas.conf配置等

1.生成访问Kafka集群的keytab文件,在Kerberos所在服务上执行如下命令

代码语言:txt
复制
[root@ip-172-31-22-86 kafkatest]# pwd
/home/ec2-user/kafkatest
[root@ip-172-31-22-86 kafkatest]# kadmin.local
Authenticating as principal hdfs/admin@CLOUDERA.COM with password.
kadmin.local:  xst -norandkey -k fayson.keytab fayson@CLOUDERA.COM

可以看到在当前目录下生成了fayson@CLOUDERA.COM账号的keytab文件。

2.创建jaas.conf文件,文件内容如下

代码语言:txt
复制
KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   keyTab="/keytab/fayson.keytab"
   principal="fayson@CLOUDERA.COM";
};

3.将keytab文件和jaas.conf文件拷贝至所有Flume Agent运行的节点

这里我们将上面的配置文件拷贝放在Flume Agent节点的/flume-keytab目录下

4.修改目录文件属主,确保flume用户有权限访问

代码语言:txt
复制
[ec2-user@ip-172-31-21-45 flume-keytab]$ sudo chown -R flume. /flume-keytab/
[ec2-user@ip-172-31-21-45 flume-keytab]$ sudo chmod -R 755 /flume-keytab/

4.配置Flume Agent


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

代码语言:txt
复制
kafka.channels = c1
kafka.sources = s1
kafka.sinks = k1

kafka.sources.s1.type =org.apache.flume.source.kafka.KafkaSource
kafka.sources.s1.kafka.bootstrap.servers =ip-172-31-26-80.ap-southeast-1.compute.internal:9092,ip-172-31-21-45.ap-southeast-1.compute.internal:9092,  ip-172-31-26-102.ap-southeast-1.compute.internal:9092
kafka.sources.s1.kafka.topics = test4
kafka.sources.s1.kafka.consumer.group.id =flume-consumer
kafka.sources.s1.kafka.consumer.security.protocol= SASL_PLAINTEXT
kafka.sources.s1.kafka.consumer.sasl.mechanism= GSSAPI
kafka.sources.s1.kafka.consumer.sasl.kerberos.service.name= kafka
kafka.sources.s1.channels = c1

kafka.channels.c1.type = memory

kafka.sinks.k1.type = hdfs
kafka.sinks.k1.channel = c1
kafka.sinks.k1.hdfs.kerberosKeytab= /flume-keytab/fayson.keytab
kafka.sinks.k1.hdfs.kerberosPrincipal= fayson@CLOUDERA.COM
kafka.sinks.k1.hdfs.path =/tmp/kafka-test
kafka.sinks.k1.hdfs.filePrefix = events-
kafka.sinks.k1.hdfs.writeFormat = Text

关于HDFS Sink的更多配置可以参考:http://flume.apache.org/FlumeUserGuide.html#hdfs-sink

2.增加Flume Agent启动参数

-Djava.security.auth.login.config=/flume-keytab/jaas.conf

配置完成后保存更改并重启FlumeAgent服务。

5.Java生产消息


1.编写jaas.conf文件

代码语言:txt
复制
KafkaClient {
  com.sun.security.auth.module.Krb5LoginModule required
  useKeyTab=true
  keyTab="/home/ec2-user/run-kafka/conf/fayson.keytab"
  principal="fayson@CLOUDERA.COM";
};

2.使用Java编写消息生产代码

代码语言:txt
复制
package com.cloudera;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.File;
import java.util.Properties;

/**
 * package: com.cloudera
 * describe: TODO
 * creat_user: Fayson
 * email: htechinfo@163.com
 * creat_date: 2017/12/12
 * creat_time: 下午3:35
 * 公众号:Hadoop实操
 */
public class ProducerTest {

    public static String TOPIC_NAME = "test4";

    public static String confPath = System.getProperty("user.dir") + File.separator + "conf";

    public static void main(String[] args) {
        try {
            String krb5conf = confPath + File.separator + "krb5.conf";
            String jaasconf = confPath + File.separator + "jaas.conf";

            System.setProperty("java.security.krb5.conf", krb5conf);
            System.setProperty("java.security.auth.login.config", jaasconf);
            System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");
//            System.setProperty("sun.security.krb5.debug", "true"); //Kerberos Debug模式

            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "ip-172-31-21-45.ap-southeast-1.compute.internal:9092,ip-172-31-26-102.ap-southeast-1.compute.internal:9020,ip-172-31-26-80.ap-southeast-1.compute.internal:9020");
            props.put(ProducerConfig.ACKS_CONFIG, "all");
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
            props.put("security.protocol", "SASL_PLAINTEXT");
            props.put("sasl.kerberos.service.name", "kafka");

            Producer<String, String> producer = new KafkaProducer<String, String>(props);
            for (int i = 0; i < 10; i++) {
                String message = i + "\t" + "fayson" + i  + "\t" + 22+i;
                ProducerRecord record = new ProducerRecord<String, String>(TOPIC_NAME, message);
                producer.send(record);
                System.out.println(message);
            }

            producer.flush();
            producer.close();

        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

3.将工程编译打包kafka-demo-1.0-SNAPSHOT.jar

代码语言:txt
复制
mvn clean package

4.使用mvn命令将工程依赖库导出

代码语言:txt
复制
mvn dependency:copy-dependencies -DoutputDirectory=/Users/fayson/Desktop/lib

将导出的jar包放在run-kafka/lib目录下。

5.编写run.sh脚本,运行测试jar包

代码语言:txt
复制
#!/bin/bash

JAVA_HOME=/usr/java/jdk1.8.0_131-cloudera

for file in `ls lib/*jar`
do
    CLASSPATH=$CLASSPATH:$file
done

export CLASSPATH

${JAVA_HOME}/bin/java com.cloudera.ProducerTest

6.conf目录文件

fayson.keytab:fayson的keytab文件

jaas.conf:java访问Kerberos环境下的配置

krb5.conf:集群的krb5配置文件

6.Kafka->Flume->HDFS流程测试


1.将第5章开发好的示例放在集群的服务器上

2.执行run.sh

代码语言:txt
复制
[ec2-user@ip-172-31-22-86 run-kafka]$ sh run.sh 

3.查看HDFS的/extwarehouse/student目录下数据

这里可以看到数据已写入HDFS指定的目录。

为天地立心,为生民立命,为往圣继绝学,为万世开太平。

温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看

本文参与 腾讯云自媒体同步曝光计划,分享自微信公众号。
原始发表:2017-12-15,如有侵权请联系 cloudcommunity@tencent.com 删除

本文分享自 Hadoop实操 微信公众号,前往查看

如有侵权,请联系 cloudcommunity@tencent.com 删除。

本文参与 腾讯云自媒体同步曝光计划  ,欢迎热爱写作的你一起参与!

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
相关产品与服务
大数据
全栈大数据产品,面向海量数据场景,帮助您 “智理无数,心中有数”!
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档