Kafka的分区策略是按照用户自定义的key字段，计算hashcode，按计算结果将该条日志存储到相应编号的分区中。
举个例子，比如DB2的一条日志：
2015-06-18-22.24.52.052782+480 I2676E403 LEVEL: Warning
PID : 14103 TID : 139828759111456 PROC : db2start
INSTANCE: db2inst1 NODE : 000
HOSTNAME: X
FUNCTION: DB2 UDB, base sys utilities, sqleGetStStLockFile, probe:14843
MESSAGE : Obtained exclusive mode lock on the file:
DATA #1 : String, 45 bytes /db2fs/home/db2inst1/sqllib/ctrl/db2strst.lck
1.用logstash解析的话，filter组件中抽取出processid字段
grok{
match=>{"message"=>"PID\s*:\s*%{NUMBER:processid}\s*TID"}
}
然后在output组件的kafka设置中指定该字段为key：
kafka {
topic_id=>"cluster"
broker_list=>"X.X.X.X:9092"
partition_key_format => "%{processid}"
}
这样logstash向kafka push数据的话kafka会对processid字段进行hashcode计算，源代码在core/src/main/scala/kafka/producer/DefaultPartitioner.scala
class DefaultPartitioner(props: VerifiableProperties = null) extends Partitioner {
private val random = new
java.util.Random
def partition(key: Any, numPartitions: Int): Int = {
Utils.abs(key.hashCode) % numPartitions
}
}
如果名为“cluster”的topic有两个分区partition0 partition1，针对processid=14103进行partition，计算过程等同于
int h=0;
char val[] = {'1','4','1','0','3'};
if (h == 0 && val.length > 0) {
for (int i = 0; i < val.length; i++) {
h = 31 * h + val[i];
}
int hash = h%2;
}
结果为1,那上面的这条日志就会存储在partition1中。
可以通过命令bin/kafka-simple-consumer-shell.sh --broker-list X.X.X.X:9092 --topic cluster --partition 0 查看分区中的数据
2.使用flume的道理一样，kafka识别header中的key字段，flume在定义interceptor时把该拦截器的name指定为"key"即可。比如：
agent.sources.source1.interceptors.i17.type=regex_extractor
agent.sources.source1.interceptors.i17.regex =PID\\s+:\\s+([0-9]{1,5})\\s+TID
agent.sources.source1.interceptors.i17.serializers=s1
agent.sources.source1.interceptors.i17.serializers.s1.name=key
根据kafka 的设计，一个partition只能由一个group中的一个consumer取数据，当我们的解析agent数量小于partition数量时，partition会在agent上做分布，当相等时，agent与partition一一对应，当agent数量大于partition时，多出的agent不工作。