草庐IT

kafka不丢数据方案、数据重复处理、数据乱序问题处理

万里长江横渡 2023-07-11 原文

一Kafka不丢数据方案

kafka处理数据不丢失，主要分为producer角度、broker角度、consumer角度
**1、【producer角度】**设置合适的ACK
Ack = 0
相当于异步发送，消息发送完毕即offset增加，继续生产。
Ack = 1
leader收到leader replica 对一个消息的接受ack才增加offset，然后继续生产。
Ack = -1
leader收到所有replica 对一个消息的接受ack才增加offset，然后继续生产。
ack在生产者指定，不同生产者可以不同。
ack设为-1，需要ISR里的所有follower应答，想要真正不丢数据，需要配合参数：
min.insync.replicas: ack为-1时生效，ISR里应答的最小follower数量。
默认为1（leader本身也算一个！），所以当ISR里除了leader本身，没有其他的follower，即使ack设为-1，相当于1的效果，不能保证不丢数据。
需要将min.insync.replicas设置大于等于2，才能保证有其他副本同步到数据。
retries = Integer.MAX_VALUE，无限重试。如果上述两个条件不满足，写入一直失败，就会无限次重试，保证数据必须成功的发送给两个副本，如果做不到，就不停的重试，除非是面向金融级的场景，面向企业大客户，或者是广告计费，跟钱的计算相关的场景下，才会通过严格配置保证数据绝对不丢失
kafka-topics.sh --bootstrap-server hadoop1:9092 --create --topic testisr2 --replication-factor 3 --partitions 4 --config min.insync.replicas=2

【producer端总结】
完全不丢结论：ack=-1 + min.insync.replicas>=2 +无限重试

2、【broker角度】
副本数大于1
min.insync.replicas大于1

3、【consumer角度】
手动提交offset，flink结合checkpoint

二、Kafka数据重复

产生的原因：发生重试造成的重复。
幂等性 + ack-1 + 事务
Kafka数据重复，可以在下一级：SparkStreaming、redis、Flink或者Hive中dwd层去重，去重的手段：分组、按照id开窗只取第一个值；

Kafka幂等性原理（单分区单会话）：producer重试引起的乱序和重复

1、重复问题的解决：

1）Kafka增加了pid和seq。Producer中每个RecordBatch都有一个单调递增的seq; Broker上每个topic的partition也会维护pid-seq的映射，并且每Commit都会更新lastSeq。

2）recordBatch到来时，broker会先检查RecordBatch再保存数据：
如果batch中 baseSeq（第一条消息的seq）比Broker维护的序号（lastSeq）大1，则保存数据，否则不保存。

三、消息乱序问题解决方案

假设我们有5个请求，batch1、batch2、batch3、batch4、batch5；
如果只有batch2 ack failed，3、4、5都保存了，那2将会随下次batch重发而造成重复。
可以设置max.in.flight.requests.per.connection=1（客户端在单个连接上能够发送的未响应请求的个数）来解决乱序，但降低了系统吞吐。
新版本kafka设置enable.idempotence=true后能够动态调整max-in-flight-request。正常情况下max.in.flight.requests.per.connection大于1。当重试请求到来时，batch 会根据 seq重新添加到队列的合适位置，并把max.in.flight.requests.per.connection设为1，这样它前面的 batch序号都比它小，只有前面的都发完了，它才能发。

kafka数据xffxff0cxff0大数据分布式

有关kafka不丢数据方案、数据重复处理、数据乱序问题处理的更多相关文章

  1. ruby - 在 64 位 Snow Leopard 上使用 rvm、postgres 9.0、ruby 1.9.2-p136 安装 pg gem 时出现问题 - 2

    我想为Heroku构建一个Rails3应用程序。他们使用Postgres作为他们的数据库，所以我通过MacPorts安装了postgres9.0。现在我需要一个postgresgem并且共识是出于性能原因你想要pggem。但是我对我得到的错误感到非常困惑当我尝试在rvm下通过geminstall安装pg时。我已经非常明确地指定了所有postgres目录的位置可以找到但仍然无法完成安装:$envARCHFLAGS='-archx86_64'geminstallpg--\--with-pg-config=/opt/local/var/db/postgresql90/defaultdb/po

  2. ruby - 解析 RDFa、微数据等的最佳方式是什么，使用统一的模式/词汇(例如 schema.org)存储和显示信息 - 2

    我主要使用Ruby来执行此操作，但到目前为止我的攻击计划如下:使用gemsrdf、rdf-rdfa和rdf-microdata或mida来解析给定任何URI的数据。我认为最好映射到像schema.org这样的统一模式，例如使用这个yaml文件，它试图描述数据词汇表和opengraph到schema.org之间的转换:#SchemaXtoschema.orgconversion#data-vocabularyDV:name:namestreet-address:streetAddressregion:addressRegionlocality:addressLocalityphoto:i

  3. ruby - 通过 rvm 升级 ruby​​gems 的问题 - 2

    尝试通过RVM将RubyGems升级到版本1.8.10并出现此错误:$rvmrubygemslatestRemovingoldRubygemsfiles...Installingrubygems-1.8.10forruby-1.9.2-p180...ERROR:Errorrunning'GEM_PATH="/Users/foo/.rvm/gems/ruby-1.9.2-p180:/Users/foo/.rvm/gems/ruby-1.9.2-p180@global:/Users/foo/.rvm/gems/ruby-1.9.2-p180:/Users/foo/.rvm/gems/rub

  4. ruby - 如何指定 Rack 处理程序 - 2

    Rackup通过Rack的默认处理程序成功运行任何Rack应用程序。例如:classRackAppdefcall(environment)['200',{'Content-Type'=>'text/html'},["Helloworld"]]endendrunRackApp.new但是当最后一行更改为使用Rack的内置CGI处理程序时，rackup给出“NoMethodErrorat/undefinedmethod`call'fornil:NilClass”:Rack::Handler::CGI.runRackApp.newRack的其他内置处理程序也提出了同样的反对意见。例如Rack

  5. ruby - 在 jRuby 中使用 'fork' 生成进程的替代方案？ - 2

    在MRIRuby中我可以这样做:deftransferinternal_server=self.init_serverpid=forkdointernal_server.runend#Maketheserverprocessrunindependently.Process.detach(pid)internal_client=self.init_client#Dootherstuffwithconnectingtointernal_server...internal_client.post('somedata')ensure#KillserverProcess.kill('KILL',

  6. ruby - 通过 RVM (OSX Mountain Lion) 安装 Ruby 2.0.0-p247 时遇到问题 - 2

    我的最终目标是安装当前版本的RubyonRails。我在OSXMountainLion上运行。到目前为止，这是我的过程:已安装的RVM$\curl-Lhttps://get.rvm.io|bash-sstable检查已知(我假设已批准)安装$rvmlistknown我看到当前的稳定版本可用[ruby-]2.0.0[-p247]输入命令安装$rvminstall2.0.0-p247注意:我也试过这些安装命令$rvminstallruby-2.0.0-p247$rvminstallruby=2.0.0-p247我很快就无处可去了。结果:$rvminstall2.0.0-p247Search

  7. ruby - Fast-stemmer 安装问题 - 2

    由于fast-stemmer的问题，我很难安装我想要的任何ruby​​gem。我把我得到的错误放在下面。Buildingnativeextensions.Thiscouldtakeawhile...ERROR:Errorinstallingfast-stemmer:ERROR:Failedtobuildgemnativeextension./System/Library/Frameworks/Ruby.framework/Versions/2.0/usr/bin/rubyextconf.rbcreatingMakefilemake"DESTDIR="cleanmake"DESTDIR=

  8. ruby - 安装 Ruby 时遇到问题(无法下载资源 "readline--patch") - 2

    当我尝试安装Ruby时遇到此错误。我试过查看this和this但无济于事➜~brewinstallrubyWarning:YouareusingOSX10.12.Wedonotprovidesupportforthispre-releaseversion.Youmayencounterbuildfailuresorotherbreakages.Pleasecreatepull-requestsinsteadoffilingissues.==>Installingdependenciesforruby:readline,libyaml,makedepend==>Installingrub

  9. ruby - Ruby 有 `Pair` 数据类型吗？ - 2

    有时我需要处理键/值数据。我不喜欢使用数组，因为它们在大小上没有限制(很容易不小心添加超过2个项目，而且您最终需要稍后验证大小)。此外，0和1的索引变成了魔数(MagicNumber)，并且在传达含义方面做得很差(“当我说0时，我的意思是head...”)。散列也不合适，因为可能会不小心添加额外的条目。我写了下面的类来解决这个问题:classPairattr_accessor:head,:taildefinitialize(h,t)@head,@tail=h,tendend它工作得很好并且解决了问题，但我很想知道:Ruby标准库是否已经带有这样一个类？ 最佳

  10. java - 从 JRuby 调用 Java 类的问题 - 2

    我正在尝试使用boilerpipe来自JRuby。我看过guide从JRuby调用Java，并成功地将它与另一个Java包一起使用，但无法弄清楚为什么同样的东西不能用于boilerpipe。我正在尝试基本上从JRuby中执行与此Java等效的操作:URLurl=newURL("http://www.example.com/some-location/index.html");Stringtext=ArticleExtractor.INSTANCE.getText(url);在JRuby中试过这个:require'java'url=java.net.URL.new("http://www

随机推荐