有没有一种方法可以创建具有缓冲区限制的Kinesis使用者?像here:
#Flush when buffer exceeds 100000 Amazon Kinesis records, 64 MB size limit or when time since last buffer exceeds 1 hour
bufferByteSizeLimit = 67108864
bufferRecordCountLimit = 100000
bufferMillisecondsLimit = 3600000
基本上,我只想在有大量数据时才启动IRecordProcessor
。我无法使用上面的连接器代码,因为我需要latest的amazon-kinesis-client
版本。
我最终实现了自己的解决方案。
ConcurrentHashMap
以存储流数据 private val recsMap = new ConcurrentHashMap[String, List[RecordStore]]
private val currByteSize = new AtomicLong(0L)
private val currRecordCount = new AtomicLong(0L)
private val currSeconds = new AtomicLong(0L)
recsMap.foreach(write2File())
// clean up
recsMap.remove(writtenRecs())
// reset counters
currByteSize.getAndSet(value)
currRecordCount.getAndSet(value)
currSeconds.getAndSet(value)