Add deduplication logic for kafka sink in case of spark task retries
Chưa có ai nhận issue này.
Đánh giá
- Độ khó
- 5/5
- Thời gian dự kiến
- Hơn một tuần
- Mức phù hợp với người mới
- 25/100
- Loại issue
- Tính năng
- Độ rõ ràng
- Khá rõ ràng
- Mức độ hoạt động
- Đình trệ
- Công nghệ
- kafka, scala
- Lĩnh vực
- stream-processing
Hướng nghiên cứu
Bắt đầu bằng cách xem xét DeduplicateKafkaSinkTransformer hiện có và số lần thử của tác vụ Spark có sẵn trong quá trình retries. Đánh giá một Kafka ProducerInterceptor xử lý retries giữa các executor và duy trì tra cứu trong bộ nhớ cho các retries trên cùng executor; được coi là hoàn thành khi các bản sao trùng lặp bị loại bỏ hoặc được chuyển hướng đến một trash topic theo các giả định đã nêu.
Do mô hình lập chỉ mục viết ra từ nội dung của issue.
Mô tả
Problem description
Spark does not provide an exactly-once behaviour for the Kafka sink, but only at-least-once, and will probably never do so (https://github.com/apache/spark/pull/25618). Under certain assumptions (no concurrent producers, only 1 destination topic, not too big micro-batches, messages don't change between retries), idempotency can still be achieved. See #177.
The DeduplicateKafkaSinkTransformer only addresses retries on an application level. However, retries (and therefore duplicates) may happen on lower levels as well, namely:
- Retry of a
DataWritingSparkTask(the Spark task that will invoke the KafkaProducer) - Internal retry of the
KafkaProducer(when encountering aRetriableException, e.g. server disconnected)
Duplicates due to an internal retry of theKafkaProducercan be prevented by settingacks=allandenable.idempotenceon the kafka writer. However, this does not take into account retries of theDataWritingSparkTaskwhich invokes theKafkaProducer. For example, if there is an intermittentTopicAuthorizationException, theKafkaProducerwill fail and not retry, but theDataWritingSparkTaskwill retry nevertheless. In such a case, duplications are still possible. Another example is executor failure due to exceeding memory limits. If an executor exceeds memory limits during theDataWritingSparkTask, it will be terminated and another executor will retry the task, which may again lead to duplicates on the destination topic.
One solution is to switch off spark task retries, but obviously, this causes other problems.
Solution
It might be possible to implement a org.apache.kafka.clients.producer.ProducerInterceptor which would deduplicate retries messages in a similar way as the DeduplicateKafkaSinkTransformer. This would capture duplicates in a scenario where the retry happens on a different executor than the original executor (memory exceeded scenario). In addition, the interceptor should keep a lookup set in memory to capture duplicates that occurred due to retries on the same executor
(TopicAuthorizationException scenario).
A reattempt could be recognized through the attempt nr of the Spark task.
If messages cannot be dropped through the ProducerInterceptor, they could at least be redirected to a trash-topic
- Ngôn ngữ chính
- Scala
- Star
- 47
- Fork
- 14
- Chỉ số merge pull request
- Không có pull request nào được merge trong 30 ngày
Hướng dẫn đóng góp
Chưa lập chỉ mục được hướng dẫn đóng góp cho kho mã nguồn này
Bắt đầu từ đâu
- Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
- Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
- Fork repository và làm thay đổi trên một nhánh.
- Mở pull request có tham chiếu số hiệu của issue.
Issue khác của AbsaOSS/hyperdrive
-
Độ khó 2/5 1-3 giờ Mức phù hợp với người mới 45/100
AbsaOSS/hyperdrive#269 ·
-
enhancement
Độ khó 3/5 1-2 ngày Mức phù hợp với người mới 42/100
AbsaOSS/hyperdrive#239 ·
-
bug
Độ khó 3/5 1-2 ngày Mức phù hợp với người mới 35/100
AbsaOSS/hyperdrive#230 ·
-
Atum integration Đang mở
Độ khó 5/5 Hơn một tuần Mức phù hợp với người mới 20/100
AbsaOSS/hyperdrive#211 · 1 bình luận · 1 reaction · 1 người được giao ·
-
internal-task
Độ khó 4/5 3-5 ngày Mức phù hợp với người mới 28/100
AbsaOSS/hyperdrive#148 · 1 người được giao ·
Tất cả issue của AbsaOSS/hyperdrive
Issue tương tự
-
RS_PixelAsPoint raises out of grid where RS_PixelAsCentroid and RS_PixelAsPolygon extrapolate Đang mở
Độ khó 2/5 1-3 giờ Mức phù hợp với người mới 75/100
-
Area: Excel support
Độ khó 2/5 1-3 giờ Mức phù hợp với người mới 75/100
orbeon/orbeon-forms#7893 ·
-
x:action/fix x:module/practice-exercise x:size/small
Độ khó 2/5 1-3 giờ Mức phù hợp với người mới 75/100
-
Độ khó 2/5 1-3 giờ Mức phù hợp với người mới 75/100
lichess-org/lila#21788 · 2 bình luận ·
-
area:Iceberg area:writer documentation requires-triage
Độ khó 2/5 1-3 giờ Mức phù hợp với người mới 75/100
apache/datafusion-comet#6147 ·