[
https://issues.apache.org/jira/browse/FLINK-3211?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15095775#comment-15095775
]
Tzu-Li (Gordon) Tai commented on FLINK-3211:
--------------------------------------------
Some issues on the originally planned implementation:
1. It is not possible to allow the user to configure start reading streams from
a specific record sequence number.
The term "sequence number" used by Kinesis is actually quite misleading. I
found out that the sequence number of each record is encoded to a specific
shard, therefore only meaningful within a shard (like offsets in Kafka
partitions). Therefore, this feature is only viable if the user specifies the
record sequence number they want to start reading for each single shard. I may
be wrong, but I doubt the usability of this feature.
2. Data ordering will be severely skewed if we start reading a stream in
parallel from the earliest record sequence number. Affect on Flink's Event Time
functionality?
Likewise, since the sequence number of records is completely independent across
partitions in the stream, there is no "stream-global" earliest record sequence
number. Therefore, when the user opts to "read from stream starting from
earliest sequence number record", what we will be doing is actually start
parallel read from each shard's earliest record possible.
Obviously, the ordering of consumed data will most likely be very out of order
with respect to the order they were produced to the stream (this is actually an
inevitable situation for distributed message queues like Kafka / Kinesis). This
reading mode is called TRIM_HORIZON in Kinesis, and the AWS KCL actually shows
the same issue. They don't guarantee consumption ordering when you have more
than one partition.
My problem is that I'm not sure how this will affect Flink's Event Time
functionality. More than often, Flink users will not be aware of this obscure
stream consumption behaviour which may have a big impact on the outputs of the
dataflow which they may not realize.
On the other hand, opting to start stream reading from the latest record
sequence number on each shard will not pose that much off an impact on the 2nd
issue. Although there is still no strict guarantee on global consumption
ordering across partitions, the ordering won't be that skewed compared to
parallel reading starting from historical points since data records are
consumed as soon as they were produced to the stream right from the beginning
of the consumer tasks.
Conclusions:
Sorry about the chunky descriptions. I felt I needed to elaborate the
considerations as detailed as possible :P
Firstly, I don't think it is required to implement opting to read from specific
sequence number. Doesn't make sense unless user specifies sequence number for
every shard. It's still do-able, so please tell me if you think this is
required.
Secondly, opting to start reading from historical points (either from the
earliest or specific sequence number) will most likely result in very skewed
consumption ordering. I'm curious about the affects of this on Flink
streaming's time-related functions. Or is this something we shouldn't be
worrying about?
> Add AWS Kinesis streaming connector
> -----------------------------------
>
> Key: FLINK-3211
> URL: https://issues.apache.org/jira/browse/FLINK-3211
> Project: Flink
> Issue Type: New Feature
> Components: Streaming Connectors
> Affects Versions: 1.0.0
> Reporter: Tzu-Li (Gordon) Tai
> Assignee: Tzu-Li (Gordon) Tai
> Original Estimate: 336h
> Remaining Estimate: 336h
>
> AWS Kinesis is a widely adopted message queue used by AWS users, much like a
> cloud service version of Apache Kafka. Support for AWS Kinesis will be a
> great addition to the handful of Flink's streaming connectors to external
> systems and a great reach out to the AWS community.
> AWS supports two different ways to consume Kinesis data: with the low-level
> AWS SDK [1], or with the high-level KCL (Kinesis Client Library) [2]. AWS SDK
> can be used to consume Kinesis data, including stream read beginning from a
> specific offset (or "record sequence number" in Kinesis terminology). On the
> other hand, AWS officially recommends using KCL, which offers a higher-level
> of abstraction that also comes with checkpointing and failure recovery by
> using a KCL-managed AWS DynamoDB "leash table" as the checkpoint state
> storage.
> However, KCL is essentially a stream processing library that wraps all the
> partition-to-task (or "shard" in Kinesis terminology) determination and
> checkpointing to allow the user to focus only on streaming application logic.
> This leads to the understanding that we can not use the KCL to implement the
> Kinesis streaming connector if we are aiming for a deep integration of Flink
> with Kinesis that provides exactly once guarantees (KCL promises only
> at-least-once). Therefore, AWS SDK will be the way to go for the
> implementation of this feature.
> With the ability to read from specific offsets, and also the fact that
> Kinesis and Kafka share a lot of similarities, the basic principles of the
> implementation of Flink's Kinesis streaming connector will very much resemble
> the Kafka connector. We can basically follow the outlines described in
> [~StephanEwen]'s description [3] and [~rmetzger]'s Kafka connector
> implementation [4]. A few tweaks due to some of Kinesis v.s. Kafka
> differences is described as following:
> 1. While the Kafka connector can support reading from multiple topics, I
> currently don't think this is a good idea for Kinesis streams (a Kinesis
> Stream is logically equivalent to a Kafka topic). Kinesis streams can exist
> in different AWS regions, and each Kinesis stream under the same AWS user
> account may have completely independent access settings with different
> authorization keys. Overall, a Kinesis stream feels like a much more
> consolidated resource compared to Kafka topics. It would be great to hear
> more thoughts on this part.
> 2. While Kafka has brokers that can hold multiple partitions, the only
> partitioning abstraction for AWS Kinesis is "shards". Therefore, in contrast
> to the Kafka connector having per broker connections where the connections
> can handle multiple Kafka partitions, the Kinesis connector will only need to
> have simple per shard connections.
> 3. Kinesis itself does not support committing offsets back to Kinesis. If we
> were to implement this feature like the Kafka connector with Kafka / ZK to
> sync outside view of progress, we probably could use ZK or DynamoDB like the
> way KCL works. More thoughts on this part will be very helpful too.
> As for the Kinesis Sink, it should be possible to use the AWS KPL (Kinesis
> Producer Library) [5]. However, for higher code consistency with the proposed
> Kinesis Consumer, I think it will be better to stick with the AWS SDK for the
> implementation. The implementation should be straight forward, being almost
> if not completely the same as the Kafka sink.
> References:
> [1]
> http://docs.aws.amazon.com/kinesis/latest/dev/developing-consumers-with-sdk.html
> [2]
> http://docs.aws.amazon.com/kinesis/latest/dev/developing-consumers-with-kcl.html
> [3]
> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Kinesis-Connector-td2872.html
> [4] http://data-artisans.com/kafka-flink-a-practical-how-to/
> [5]
> http://docs.aws.amazon.com//kinesis/latest/dev/developing-producers-with-kpl.html#d0e4998
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
