微信公众号：大数据开发运维架构
关注可了解更多大数据相关的资讯。问题或建议，请公众号留言;
如果您觉得“大数据开发运维架构”对你有帮助，欢迎转发朋友圈
从微信公众号拷贝过来，格式有些错乱，建议直接去公众号阅读
当kafka开启Kerberos认证后，如何使用Flink生产或消费数据呢？其实就是在生产消费者的代码中加入jaas.conf、keytab这些认证有关的配置，下面我们直接看代码：
版本信息：
flink1.9.0
kafka0.10.0
这里提示一下，如果版本依赖的不一致会报错，一定要对应版本：
java.lang.NoSuchMethodError:org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread
1.其实连接Kerberos集群很简单，需要下面三个文件：
1).KerberosServer的配置文件krb5.conf,让程序知道我应该哪个kdc去登录认证；
[libdefaults]udp_preference_limit=1 renew_lifetime=3650dforwardable=truedefault_realm=CHINAUNICOMticket_lifetime=3650ddns_lookup_realm=falsedns_lookup_kdc=falsedefault_ccache_name=/tmp/krb5cc_%{uid} #default_tgs_enctypes = aes des3-cbc-sha1 rc4 des-cbc-md5 #default_tkt_enctypes = aes des3-cbc-sha1 rc4 des-cbc-md5[domain_realm] .CHINAUNICOM = CHINAUNICOM[logging]default=FILE:/var/log/krb5kdc.logadmin_server=FILE:/var/log/kadmind.log kdc = FILE:/var/log/krb5kdc.log[realms]CHINAUNICOM={ admin_server = master98.hadoop.ljskdc=master98.hadoop.ljs }
2).认证肯定需要指定认证方式这里需要一个jaas.conf文件，一般集群的conf目录下都有；
KafkaClient{com.sun.security.auth.module.Krb5LoginModulerequireduseKeyTab=truekeyTab="D:\\kafkaSSL\\kafka.service.keytab"storeKey=trueuseTicketCache=falseprincipal="kafka/salver32.hadoop.unicom@CHINAUNICOM"serviceName=kafka;};
3).就是用户的登录认证票据和认证文件,票据和keytab文件这里就不在贴了；
2.为防止你依赖报错，这里贴下pom.xml依赖，可能有些冗余，自己删除即可：
org.apache.kafkakafka-clients${kafka.version}compileorg.apache.flinkflink-hadoop-fs${flink.version}org.apache.hadoophadoop-common${hadoop.version}org.apache.hadoophadoop-hdfs${hadoop.version}org.apache.httpcomponentshttpclient${httpclient.version}org.apache.flinkflink-connector-kafka-0.10_2.111.9.0compile
4.Flink接收socket端消息，发送到kafka:
5.Flink将socket接收的数据发送Kafka,代码实例：
packagecom.hadoop.ljs.flink.streaming;importcom.hadoop.ljs.flink.utils.CustomKeyedSerializationSchema;importorg.apache.flink.streaming.api.datastream.DataStream;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;importorg.apache.kafka.clients.producer.ProducerConfig;importjava.util.Properties;/***@author: Created By lujisen*@companyChinaUnicom Software JiNan*@date: 2020-02-29 09:31*@version: v1.0*@description: com.hadoop.ljs.flink.streaming */publicclassFlinkKafkaKerberosProducer{publicstaticfinalString topic="topic1";publicstaticfinalString krb5Conf="D:\\kafkaSSL\\krb5.conf";publicstaticfinalString kafkaJaasConf="D:\\kafkaSSL\\kafka_client_jaas.conf";publicstaticfinalString bootstrapServers="salver31.hadoop.unicom:6667,salver32.hadoop.unicom:6667";publicstaticfinalString hostname="localhost";publicstaticfinalintport=9000;publicstaticvoidmain(String[] args)throwsException{//在windows中设置JAAS，也可以通过-D方式传入System.setProperty("java.security.krb5.conf", krb5Conf);System.setProperty("java.security.auth.login.config", kafkaJaasConf);/*获取flink流式计算执行环境*/finalStreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();/*从Socket端接收数据*/DataStream dataSource = senv.socketTextStream(hostname, port,"\n");/*下面可以根据自己的需求进行自动的转换*//*接收的数据，中间可经过复杂的处理，最后发送到kafka端*/dataSource.addSink(newFlinkKafkaProducer010(topic,newCustomKeyedSerializationSchema(), getProducerProperties()));/*启动*/senv.execute("FlinkKafkaProducer"); }publicstaticPropertiesgetProducerProperties(){Properties props =newProperties();props.put("bootstrap.servers", bootstrapServers);props.put("acks","1");props.put("retries",3);props.put("batch.size",16384);props.put("linger.ms",1);props.put("buffer.memory",33554432);props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.ByteArraySerializer");props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.ByteArraySerializer");props.put("security.protocol","SASL_PLAINTEXT");props.put("sasl.kerberos.service.name","kafka");props.put("sasl.mechanism","GSSAPI");returnprops; }}
6.Flink连接kafka消费消息，代码实例：
package com.hadoop.ljs.flink.streaming;importcom.hadoop.ljs.flink.utils.KafkaCommonRecordSchema;importorg.apache.flink.api.common.functions.MapFunction;importorg.apache.flink.api.common.serialization.SimpleStringSchema;importorg.apache.flink.streaming.api.datastream.DataStream;importorg.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;importorg.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;importorg.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;importorg.apache.kafka.clients.consumer.ConsumerRecord;importjava.util.HashMap;importjava.util.Map;importjava.util.Properties;/** * @author: Created By lujisen * @company ChinaUnicom Software JiNan * @date: 2020-02-29 09:31 * @version: v1.0 * @description: com.hadoop.ljs.flink.streaming */publicclassFlinkKafkaKerberosConsumer{publicstaticfinalStringkrb5Conf="D:\\kafkaSSL\\krb5.conf";publicstaticfinalStringkafkaJaasConf="D:\\kafkaSSL\\kafka_client_jaas.conf";publicstaticfinalStringtopic="topic1";publicstaticfinalStringconsumerGroup="test_topic1";publicstaticfinalStringbootstrapServer="salver31.hadoop.unicom:6667,salver32.hadoop.unicom:6667";publicstaticvoid main(String[] args)throwsException{//在windows中设置JAAS，也可以通过-D方式传入System.setProperty("java.security.krb5.conf", krb5Conf);System.setProperty("java.security.auth.login.config", kafkaJaasConf);finalStreamExecutionEnvironmentenv =StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);FlinkKafkaConsumer010 consumer010 = newFlinkKafkaConsumer010(topic,newSimpleStringSchema(), getComsumerProperties()); consumer010.setStartFromEarliest(); //source从kafkaDataStream dataStream = env.addSource(consumer010);dataStream.print();try{ env.execute();}catch(Exceptionex) { ex.printStackTrace(); } }privatestaticPropertiesgetComsumerProperties() {Propertiesprops = newProperties();props.put("bootstrap.servers",bootstrapServer);props.put("group.id",consumerGroup);props.put("auto.offset.reset","earliest");props.put("security.protocol","SASL_PLAINTEXT");props.put("sasl.kerberos.service.name","kafka");props.put("sasl.mechanism","GSSAPI");returnprops; }}
