Spark消费Kafka如何实现精准一次性消费？-程序员宅基地

技术标签： spark kafka Spark 大数据

在这里插入图片描述

1.定义

精确一次消费（Exactly-once）是指消息一定会被处理且只会被处理一次。不多不少就一次处理。

如果达不到精确一次消费，可能会达到另外两种情况：

至少一次消费（at least once），主要是保证数据不会丢失，但有可能存在数据重复问题。
最多一次消费 （at most once），主要是保证数据不会重复，但有可能存在数据丢失问题。

如果同时解决了数据丢失和数据重复的问题，那么就实现了精确一次消费的语义了。

2. 问题如何产生

数据何时会丢失：比如实时计算任务进行计算，到数据结果存盘之前，进程崩溃，假设在进程崩溃前kafka调整了偏移量，那么kafka就会认为数据已经被处理过，即使进程重启，kafka也会从新的偏移量开始，所以之前没有保存的数据就被丢失掉了。

数据何时会重复：如果数据计算结果已经存盘了，在kafka调整偏移量之前，进程崩溃，那么kafka会认为数据没有被消费，进程重启，会重新从旧的偏移量开始，那么数据就会被2次消费，又会被存盘，数据就被存了2遍，造成数据重复。

3.解决方案

方案一：利用关系型数据库的事务进行处理

出现丢失或者重复的问题，核心就是偏移量的提交与数据的保存，不是原子性的。如果能做成要么数据保存和偏移量都成功，要么两个失败。那么就不会出现丢失或者重复了。

这样的话可以把存数据和偏移量放到一个事务里。这样就做到前面的成功，如果后面做失败了，就回滚前面那么就达成了原子性。

问题与限制

数据库选型受限, 只能使用支持事务的关系型数据库，如mysql， oracle ，无法使用其他功能强大的nosql数据库。
如果保存的数据量较大一个数据库节点不够，多个节点的话，还要考虑分布式事务的问题。做分布式事务,结构复杂，拖慢性能。
如果做本地事务，只能把分区数据提取到driver中进行保存，降低并发，增加executor到driver的数据传递io。

方案二：手动提交偏移量+幂等性处理

咱们知道如果能够同时解决数据丢失和数据重复问题，就等于做到了精确一次消费。

那咱们就各个击破。

首先解决数据丢失问题，办法就是要等数据保存成功后再提交偏移量，所以就必须手工来控制偏移量的提交时机。

但是如果数据保存了，没等偏移量提交进程挂了，数据会被重复消费。怎么办？那就要把数据的保存做成幂等性保存。即同一批数据反复保存多次，数据不会翻倍，保存一次和保存一百次的效果是一样的。如果能做到这个，就达到了幂等性保存，就不用担心数据会重复了。

难点

话虽如此，在实际的开发中手动提交偏移量其实不难，难的是幂等性的保存，有的时候并不一定能保证。所以有的时候只能优先保证的数据不丢失。数据重复难以避免。即只保证了至少一次消费的语义。

文章已收录于https://github.com/wangsl123/dzblog，欢迎start。微信关注【鄙人王道长】，也可第一时间看到文章。

在这里插入图片描述

本文链接：https://blog.csdn.net/wangsl754/article/details/107479977

原作者删帖不实内容删帖广告或垃圾文章投诉

智能推荐

使用UmcFramework和unimrcpclient.xml连接多个SIP设置的配置指南及C代码示例

在多媒体通信领域，MRCP（Media Resource Control Protocol）协议被广泛用于控制语音识别和合成等媒体资源。UniMRCP是一个开源的MRCP实现，提供了客户端和服务端的库。UmcFramework是一个基于UniMRCP客户端库的示例应用程序框架，它帮助开发者快速集成和测试MRCP客户端功能。本文将详细介绍如何使用UmcFramework和unimrcpclient.xml配置文件连接到多个SIP设置，以及如何用C代码进行示例说明。

java.net.ProtocolException: Server redirected too many times (20)-程序员宅基地

文章浏览阅读3k次。报错：java.net.ProtocolException: Server redirected too many times (20)1.没有检查到cookie，一直循环重定向。解决：CookieHandler.setDefault(new CookieManager(null, CookiePolicy.ACCEPT_ALL));URL url = new URL(url); ..._java.net.protocolexception: server redirected too many times (20)

springboot启动报错 Failed to scan *****/derbyLocale_ja_JP.jar from classloader hierarchy_failed to scan from classloader hierarchy-程序员宅基地

文章浏览阅读4.1k次。问题这是部分报错信息2019-07-11 14:03:34.283 WARN [restartedMain][DirectJDKLog.java:175] - Failed to scan [file:/D:/repo/org/apache/derby/derby/10.14.2.0/derbyLocale_ja_JP.jar] from classloader hierarchyjava...._failed to scan from classloader hierarchy

MATLAB-ones函数_matlab中ones函数-程序员宅基地

文章浏览阅读2.8k次，点赞3次，收藏7次。在MATLAB中，ones函数用于创建一个指定大小的由1组成的矩阵或数组。_matlab中ones函数

解决PS等软件出现应用程序无法正常启动(0xc000007b)_photoshop应用程序无法正常启动0xc000007b。请单击“确认”关闭应用程序。-程序员宅基地

文章浏览阅读3.9w次，点赞2次，收藏9次。　　在使用电脑办公过程中，安装应用程序时难免遇到无法安装或者无法正常启动的问题，这对我们使用电脑带来了诸多不便。那遇到应用程序无法正常启动的问题要如何解决呢?相信大家肯定都是十分疑问的，每次都是只能忍痛重新安装软件。今天，小编就和大家探讨下应用程序无法正常启动的解决方法，帮助大家排忧解难。0xc000007b电脑图解1　　第一种方案：SFC检查系统完整性来尝试修复丢失文件　　1、打开电脑搜索输入cmd.exe，选择以管理员身份运行，跳出提示框时选择继续。0xc000007b电脑图解2_photoshop应用程序无法正常启动0xc000007b。请单击“确认”关闭应用程序。

oracle介质恢复和实例恢复的异同-程序员宅基地

文章浏览阅读396次。1、概念 REDO LOG是Oracle为确保已经提交的事务不会丢失而建立的一个机制。实际上REDO LOG的存在是为两种场景准备的：实例恢复（INSTANCE RECOVERY）；介质恢复（MEDIA RECOVERY）。实例恢复的目的是在数据库发生故障时，确保BUFFER CACHE中的数据不会丢失，不会造成数据库的..._oracle 实例恢复和介质恢复