1. 云图网首页
  2. 技术专区
  3. 大数据

Kafka 分区备份实战详解大数据

• 大数据

1.概述

　　在 Kafka 集群中，我们可以对每个 Topic 进行一个或是多个分区，并为该 Topic 指定备份数。这部分元数据信息都是存放在 Zookeeper 上，我们可以使用 zkCli 客户端，通过 ls 和 get 命令来查看元数据信息。通过 log.dirs 属性控制消息存放路径，每个分区对应一个文件夹，文件夹命名方式为：TopicName-PartitionIndex，该文件夹下存放这该分区的所有消息和索引文件，如下图所示：

Kafka 分区备份实战详解大数据

2.内容

　　Kafka 集群在生产消息入库的时候，通过 Key 来进行分区存储，按照相应的算法，生产分区规则，让所生产的消息按照该规则分布到不同的分区中，以达到水平扩展和负载均衡。而我们在消费这些消息的时候，可以使用多线程来消费该 Topic 下的所有分区中的消息。

　　分区规则的制定，通过实现 kafka.producer.Partitioner 接口，该接口我们可以进行重写，按照自己的方式去实现分区规则。如下，我们按照 Key 的 Hash 值，然后取模得到分区索引，代码如下所示：

package cn.hadoop.hdfs.kafka.partition; 
 
import kafka.producer.Partitioner; 
import kafka.utils.VerifiableProperties; 
 
/** 
 * @Date Nov 3, 2016 
 * 
 * @Author dengjie 
 * 
 * @Note 先 Hash 再取模，得到分区索引 
 */ 
public class CustomerPartitioner implements Partitioner { 
 
    public CustomerPartitioner(VerifiableProperties props) { 
    } 
 
    public int partition(Object key, int numPartitions) { 
        int partition = 0; 
        String k = (String) key; 
        partition = Math.abs(k.hashCode()) % numPartitions; 
        return partition; 
    } 
 
}

　　在创建 Topic 的时候，若按照上述规则创建分区，分区数最后为 Brokers 的整数倍，这样才能发挥其负载均衡的作用，比如：当前我们集群节点由 3 个 Broker 组成，如下图所示：

Kafka 分区备份实战详解大数据

2.1 创建分区

　　我们在创建分区的时候，可以通过 Kafka 提供的客户端命令进行创建，如下，我们创建一个6分区，3备份的一个 Topic，命令如下所示：

./kafka-topics.sh --create --zookeeper k1:2181,k2:2181,k3:2181 --replication-factor 3 --partitions 6 --topic ke_test

　　这里需要注意的是，指定备份数的时候，备份数要小于等于 Brokers 数。否则创建失败。在创建分区的时候，假设，我们只创建 2 个分区，而我们上述图中， Brokers 有 3 个，会造成有一个 Broker 上没有该 Topic 的分区，以致分布不均。

2.2 分区入库

　　一般，我们在入库消息的时候，都有使用 Kafka 的 API，如下，我们使用生产 API ，按照上述的 Hash 取模规则，进行分区入库，代码如下所示：

package cn.hadoop.hdfs.kafka.partition; 
 
import java.util.List; 
import java.util.Properties; 
 
import cn.hadoop.hdfs.kafka.partition.data.FileRead; 
import kafka.javaapi.producer.Producer; 
import kafka.producer.KeyedMessage; 
import kafka.producer.ProducerConfig; 
 
/** 
 * @Date Nov 3, 2016 
 * 
 * @Author dengjie 
 * 
 * @Note 按照先 Hash 再取模的规则，进行分区入库 
 */ 
public class PartitionerProducer { 
    public static void main(String[] args) { 
        producerData(); 
    } 
 
    private static void producerData() { 
        Properties props = new Properties(); 
        props.put("serializer.class", "kafka.serializer.StringEncoder"); 
        props.put("metadata.broker.list", "k1:9092,k2:9092,k3:9092"); 
        props.put("partitioner.class", "cn.hadoop.hdfs.kafka.partition.CustomerPartitioner"); 
        Producer<String, String> producer = new Producer<String, String>(new ProducerConfig(props)); 
        String topic = "ke_test"; 
        List<String> list = FileRead.readData(); 
        for (int i = 0; i < list.size(); i++) { 
            String k = "key" + i; 
            String v = new String(list.get(i)); 
            producer.send(new KeyedMessage<String, String>(topic, k, v)); 
            if (i == (list.size() - 1)) { 
                return; 
            } 
        } 
        producer.close(); 
    } 
}

　　这里，我们分析发现，生产者在生产消息入库时，会按照 CustomerPartitioner 的规则，进行分区入库，在入库时，将 Key 先做 Hash，然后分区数取模（这里分区数是 6）.我们计算可以得到一下信息：

hashCode("key0") % 6 = 1 
hashCode("key1") % 6 = 2 
hashCode("key2") % 6 = 3 
hashCode("key3") % 6 = 4 
hashCode("key4") % 6 = 5 
hashCode("key5") % 6 = 0 
// ... 以此循环

　　按照该表述规则进行分区入库。

2.3 分区入库验证

　　接下里，我们通过 Kafka 的消费者 API 来验证，在消费时，消费 Topic 各分区的详情，代码如下所示：

package cn.hadoop.hdfs.kafka.partition; 
 
import java.util.HashMap; 
import java.util.List; 
import java.util.Map; 
import java.util.Properties; 
 
import kafka.consumer.Consumer; 
import kafka.consumer.ConsumerConfig; 
import kafka.consumer.ConsumerIterator; 
import kafka.consumer.KafkaStream; 
import kafka.javaapi.consumer.ConsumerConnector; 
import kafka.message.MessageAndMetadata; 
 
/** 
 * @Date Nov 3, 2016 
 * 
 * @Author dengjie 
 * 
 * @Note 通过 Kafka 的消费者 API 验证分区入库的消息 
 */ 
public class PartitionerConsumer { 
    public static void main(String[] args) { 
        String topic = "ke_test"; 
        ConsumerConnector consumer = Consumer.createJavaConsumerConnector(createConsumerConfig()); 
        Map<String, Integer> topicCountMap = new HashMap<String, Integer>(); 
        topicCountMap.put(topic, new Integer(1)); 
        Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap); 
        KafkaStream<byte[], byte[]> stream = consumerMap.get(topic).get(0); 
        ConsumerIterator<byte[], byte[]> it = stream.iterator(); 
        while (it.hasNext()) { 
            MessageAndMetadata<byte[], byte[]> mam = it.next(); 
            System.out.println("consume: Partition [" + mam.partition() + "] Message: [" + new String(mam.message()) 
                    + "] .."); 
        } 
 
    } 
 
    private static ConsumerConfig createConsumerConfig() { 
        Properties props = new Properties(); 
        props.put("group.id", "group1"); 
        props.put("zookeeper.connect", "zk1:2181,zk2:2181,zk3:2181"); 
        props.put("zookeeper.session.timeout.ms", "40000"); 
        props.put("zookeeper.sync.time.ms", "200"); 
        props.put("auto.commit.interval.ms", "1000"); 
        props.put("auto.offset.reset", "smallest"); 
        return new ConsumerConfig(props); 
    } 
}

　　这里笔者只是验证消费数据，若在实际生产线上，需将上述单线程消费改造成多线程消费，来提升处理消息的能力。

2.4 验证结果

　　这里，我们线运行生产者，让其生产消息，并分区入库；然后，在启动消费者，消费消息验证其结果，如下图所示：

Kafka 分区备份实战详解大数据

3.总结

　　需要注意的是，分区数建议为 Brokers 的整数倍，让其达到均匀分布；备份数必须小于等于 Brokers。以及，多线程消费的控制，其线程数建议和分区数相等。

4.结束语

　　这篇博客就和大家分享到这里，如果大家在研究学习的过程当中有什么问题，可以加群进行讨论或发送邮件给我，我会尽我所能为您解答，与君共勉！

原创文章，作者：ItWorker，如若转载，请注明出处：https://blog.ytso.com/9977.html

赞 (0)
0
上一篇 2021年7月19日
下一篇 2021年7月19日

相关推荐

发表回复

登录后才能评论