Kafka topic及partition设计
1、对于银行应用日志，一个系统建一个topic，每台主机对应一个partition，规则为，flume采集时，同一个应用，数据送到同一个topic，一个主机，送一个partition，这样做是为了同一个日志的数据在一个partition中，顺序不会乱。另，flume配置文件可以配置sink的topic和partition
id（xxx.kafka.topic = xxx;dafaultPartitionId = x）。
2、flume送数据到kafka，flume配置文件配置文件
|
a1.sources = r1
a1.channels = c1
a1.sinks = k1
a1.sources.r1.channels = c1
a1.sinks.k1.channel = c1
a1.channels.c1.type = memory
a1.channels.c1.capacity = 150000
a1.channels.c1.transactionCapacity = 1500
##自定义source，见前一篇博客（flume
1.7 TailDir source重复获取数据集不释放资源解决办法）
a1.sources.r1.type = TaildirSourceSelf
a1.sources.r1.batchSize = 500
a1.sources.r1.skipToEnd = true
a1.sources.r1.ignorePathChange = true
a1.sources.r1.positionFile = test/logConf/xxx_taildir_position.json
a1.sources.r1.writePosInterval = 5000
a1.sources.r1.filegroups = f0
a1.sources.r1.filegroups.f0 = xxx/xxx/*.*
a1.sources.r1.file.header.enabled = true
a1.sources.r1.file.name.key = tag.default.filename
a1.sources.r1.dir.header.enabled = true
a1.sources.r1.dir.name.key = tag.default.dirname
a1.sources.r1.inode.header.enabled = false
a1.sources.r1.inode.key = tag.default.inode
# For Kafka Sink
a1.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
a1.sinks.k1.brokerList = mn01:9092
a1.sinks.k1.kafka.topic = test-topic
a1.sinks.k1.defaultPartitionId = 0
a1.sinks.k1.flumeBatchSize = 500
a1.sinks.k1.kafka.producer.acks = 1
a1.sinks.k1.kafka.producer.linger.ms = 1
a1.sinks.k1.kafka.producer.max.request.size = 5120000
|
3、SparkStreaming从kafka获取数据
Spark提供两种方法从kafka拿数据：
1）Receiver方式：
a、其实是通过zookeeper连接kafka队列；
b、kafka中topic的partition数目与sparkrdd中partition数目没有关系，增加topic的partition数，不会增加spark的处理并行度，仅仅是增加获取数据的Receiver；
c、Receiver方式，据都存储在Spark Executor的内存中，一旦spark停止运行（如机器崩溃），数据将无法恢复，只有则不开启WAL（write
ahead log）机制，及预写日志的方式同步将数据写到分布式系统中，虽然可以恢复，但是效率低，每份数据需要复制两份；
2）Director方式：
a、这是spark1.3引进的新方法，直接从kafka的broker分区中读数据，跳过zookeeper，也没有receiver；
b、kafka中topic的partition与spark
rdd的partition一一对应，也就是说增加kafka中topic的partition数目，就增加了spark的并行处理度；
c、该方式不会复制两份数据，因为kafka本身就有高可用，kafka会做数据备份，宕机后，可以利用kafka副本恢复；
补充：这种方式数据的offset存在spark的checkpoint中，不然每次重启spark，就会从kafka的最新offset位置读数据，会丢数据；所以，spark需要设置checkpoint，在创建JavaStreamingContext时，建议使用
JavaStreamingContext.getOrCreate(sparkChkDir,newStreamContextFunction());//第一个参数：checkpoint路径；第二个参数，返回新的JavaStreamingContext；
及如果checkpoint存在就从checkpoint得到sparkStreamingContext，不存在就创建sparkStreamingContext；
下面是我从spark获取数据的代码（java版）：
|
public static JavaDStream<String> getJavaDStreamFromKafka(String _Brokers, String _Topics, String _Metadata,
JavaStreamingContext _JavaStreamingContext) {
//返回该“，”分隔的所有topic的数据
HashSet<String> _TopicsSet = new HashSet<String>(Arrays.asList(_Topics.split(",")));
Map<String, String> _KafkaParams = new HashMap<String, String>();
_KafkaParams.put(_Metadata, _Brokers);
_KafkaParams.put("group.id", "xxxx");
_KafkaParams.put("fetch.message.max.bytes", "5120000");
// Create direct kafka stream with brokers and topics
JavaDStream<String> _MessagesLines = KafkaUtils
.createDirectStream(_JavaStreamingContext, String.class, String.class, StringDecoder.class,
StringDecoder.class, _KafkaParams, _TopicsSet)
.map(new Function<Tuple2<String, String>, String>() {
public String call(Tuple2<String, String> tuple2) {
return tuple2._2();
}
});
return _MessagesLines;
}
|