spark读取kafka数据(两种方式比较及flume配置文件)
spark读取kafka数据(两种方式比较及flume配置文件)。
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 getJavaDStreamFromKafka(String _Brokers, String _Topics, String _Metadata,
JavaStreamingContext _JavaStreamingContext) {
//返回该“,”分隔的所有topic的数据
HashSet _TopicsSet = new HashSet(Arrays.asList(_Topics.split(",")));
Map _KafkaParams = new HashMap();
_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 _MessagesLines = KafkaUtils
.createDirectStream(_JavaStreamingContext, String.class, String.class, StringDecoder.class,
StringDecoder.class, _KafkaParams, _TopicsSet)
.map(new Function, String>() {
public String call(Tuple2 tuple2) {
return tuple2._2();
}
});
return _MessagesLines;
}
上一篇: JavaEE——JavaScript
下一篇: C语言实例之困死自己的智能蛇