samza-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Jagadish Venkatraman <jagadish1...@gmail.com>
Subject Re: Questions about checkpointing and Kinesis
Date Tue, 07 Feb 2017 06:48:00 GMT
Great to hear this development on the kinesis consumer!

Let me answer some of your questions here.

*1. "Kinesis does not have a listener/push framework (unless I missed
something)"*

 Let me point out that Kinesis has both a push and a pull based model. You
can choose to implement either for your use-case.

   - *Pull:* The pull based model supports obtaining an *ShardIterator* for
   a shard and iterating on it. Please refer the docs for the
   *ShardIteratorRequest* here
   <http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/services/kinesis/model/GetShardIteratorRequest.html>.
   It's important that you are aware of re-sharding - The
   record.getNextShardIterator() can return null if there was a merge / a
   split for the shard. (You can trivially handle re-shards by re-starting)
   - *Push:* The push based model directly uses KCL (The Kinesis Client
   Library) to subscribe to events. KCL will handle sharding, re-balancing,
   checkpointing internally.

Your consumer can implement the *CheckpointListener
<https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/checkpoint/CheckpointListener.java>*
interface to get notified when Samza performs a checkpoint. You can save
the offsets of the shards to kinesis by invoking
IRecordCheckpointer.checkpoint
<https://github.com/awslabs/amazon-kinesis-client/blob/master/src/main/java/com/amazonaws/services/kinesis/clientlibrary/interfaces/IRecordProcessorCheckpointer.java>
APIs.


*2. "My plan was to use the default KafkaCheckpointManagerFactory on an
timed interval basis"*

The checkpoint manager merely provides persistence for the checkpoints. (In
that sense, it's actually a checkpoint writer). You probably don't want to
implement a custom checkpoint manager.


*3. "What exactly is being checkpointed? What value can I retrieve to use
as an offset for my Kinesis stream? Or is this something I need to keep
track of in a store? If so, what is the point of checkpointing? Can I use
RocksDb to save the Kinesis offset at every document (efficiently that
is)?"*

*- *Samza checkpoints [ssp, offset] pairs for your tasks.
- Kinesis has an implicit notion of sequence numbers for every shard in a
stream. You can use that as offsets.
- You don't want to record offsets in a separate store. If you want Samza
to manage offsets, Samza will use Kafka internally. If you want Kinesis to
manage offsets (KCL) , Kinesis will use DynamoDb to store its offsets.




On Mon, Feb 6, 2017 at 2:57 PM, Chad Greenberg <evil_goodness@hotmail.com>
wrote:

> Starting on an integration project between a Kinesis stream and Samza,
> despite have no background in either, but I am familiar with most other
> messaging/queuing systems.
>
>
> Decided to keep all state management within Samza instead of using
> Kinesis' client library. My plan was to use the default
> KafkaCheckpointManagerFactory on an timed interval basis, but I have a few
> questions.
>
>
> What exactly is being checkpointed? What value can I retrieve to use as an
> offset for my Kinesis stream? Or is this something I need to keep track of
> in a store? If so, what is the point of checkpointing? Can I use RocksDb to
> save the Kinesis offset at every document (efficiently that is)?
>
>
> Related to Kinesis and not quite Samza, it does not have a listener/push
> framework, but it requires constant polling (unless I missed something).
> First of all, I was going to have a partition for each Kinesis shard
> partition. But the next question is, should I simply have a while(true)
> polling method inside my consumer(BlockingEnvelopeMap)? Seems inefficient,
> even with a timeout. How can I get new data to instantiate a new consumer?
> My consumer will put a new document to my task.
>
>
> Cheers.
>
>
>
>


-- 
Jagadish V,
Graduate Student,
Department of Computer Science,
Stanford University

Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message