ປື້ມ "ສາຍ Kafka ໃນການປະຕິບັດ. ແອັບພລິເຄຊັນແລະບໍລິການຈຸລະພາກສໍາລັບການເຮັດວຽກໃນເວລາຈິງ"

ສະບາຍດີຊາວ Khabro! ປື້ມນີ້ແມ່ນເຫມາະສົມສໍາລັບນັກພັດທະນາຜູ້ທີ່ຕ້ອງການເຂົ້າໃຈການປຸງແຕ່ງກະທູ້. ຄວາມເຂົ້າໃຈການຂຽນໂປຼແກຼມທີ່ແຈກຢາຍຈະຊ່ວຍໃຫ້ທ່ານເຂົ້າໃຈ Kafka ແລະ Kafka Stream ໄດ້ດີຂຶ້ນ. ມັນຈະເປັນການດີທີ່ຈະຮູ້ຈັກກອບ Kafka ຕົວຂອງມັນເອງ, ແຕ່ນີ້ບໍ່ຈໍາເປັນ: ຂ້ອຍຈະບອກເຈົ້າທຸກສິ່ງທີ່ທ່ານຕ້ອງການ. ນັກພັດທະນາ Kafka ທີ່ມີປະສົບການ ແລະຈົວໃໝ່ຈະຮຽນຮູ້ວິທີສ້າງແອັບພລິເຄຊັນການປະມວນຜົນກະແສທີ່ໜ້າສົນໃຈໂດຍໃຊ້ຫ້ອງສະໝຸດ Kafka Stream ໃນປຶ້ມຫົວນີ້. ນັກພັດທະນາ Java ລະດັບປານກາງແລະຂັ້ນສູງທີ່ຄຸ້ນເຄີຍກັບແນວຄວາມຄິດເຊັ່ນ serialization ຈະຮຽນຮູ້ທີ່ຈະນໍາໃຊ້ທັກສະຂອງເຂົາເຈົ້າເພື່ອສ້າງຄໍາຮ້ອງສະຫມັກ Kafka Stream. ລະຫັດແຫຼ່ງຂອງປື້ມແມ່ນຂຽນໃນ Java 8 ແລະເຮັດໃຫ້ການນໍາໃຊ້ທີ່ສໍາຄັນຂອງ Java 8 lambda syntax, ດັ່ງນັ້ນການຮູ້ວິທີການເຮັດວຽກກັບຟັງຊັນ lambda (ແມ້ແຕ່ຢູ່ໃນພາສາການຂຽນໂປລແກລມອື່ນ) ຈະເປັນປະໂຫຍດ.

ຫຍໍ້ມາຈາກ. 5.3. ການລວບລວມແລະການດໍາເນີນການ windowing

ໃນພາກນີ້, ພວກເຮົາຈະໄປຄົ້ນຫາພາກສ່ວນທີ່ໂດດເດັ່ນທີ່ສຸດຂອງ Kafka Stream. ມາຮອດປະຈຸບັນພວກເຮົາໄດ້ກວມເອົາລັກສະນະດັ່ງຕໍ່ໄປນີ້ຂອງ Kafka Stream:

• ການສ້າງ topology ການປຸງແຕ່ງ;
• ການນໍາໃຊ້ສະຖານະໃນຄໍາຮ້ອງສະຫມັກ streaming;
• ປະຕິບັດການເຊື່ອມຕໍ່ນ້ໍາຂໍ້ມູນ;
• ຄວາມແຕກຕ່າງລະຫວ່າງການຖ່າຍທອດເຫດການ (KStream) ແລະອັບເດດສະຕຣີມ (KTable).

ໃນຕົວຢ່າງຕໍ່ໄປນີ້ພວກເຮົາຈະນໍາເອົາອົງປະກອບທັງຫມົດເຫຼົ່ານີ້ມາຮ່ວມກັນ. ນອກນັ້ນທ່ານຍັງຈະຮຽນຮູ້ກ່ຽວກັບການ windowing, ຄຸນນະສົມບັດທີ່ຍິ່ງໃຫຍ່ອີກອັນຫນຶ່ງຂອງຄໍາຮ້ອງສະຫມັກ streaming. ຕົວຢ່າງທໍາອິດຂອງພວກເຮົາຈະເປັນການລວບລວມງ່າຍດາຍ.

5.3.1. ການລວບລວມການຂາຍຫຼັກຊັບໂດຍຂະແຫນງອຸດສາຫະກໍາ

ການລວບລວມແລະການຈັດກຸ່ມແມ່ນເຄື່ອງມືທີ່ສໍາຄັນໃນເວລາທີ່ເຮັດວຽກກັບຂໍ້ມູນການຖ່າຍທອດ. ການກວດສອບບັນທຶກສ່ວນບຸກຄົນຍ້ອນວ່າພວກເຂົາໄດ້ຮັບແມ່ນມັກຈະບໍ່ພຽງພໍ. ເພື່ອສະກັດຂໍ້ມູນເພີ່ມເຕີມຈາກຂໍ້ມູນ, ມັນຈໍາເປັນຕ້ອງຈັດກຸ່ມແລະສົມທົບພວກມັນ.

ໃນຕົວຢ່າງນີ້, ທ່ານຈະໃສ່ເຄື່ອງນຸ່ງຂອງພໍ່ຄ້າມື້ຫນຶ່ງທີ່ຕ້ອງການຕິດຕາມປະລິມານການຂາຍຫຼັກຊັບຂອງບໍລິສັດໃນຫຼາຍໆອຸດສາຫະກໍາ. ໂດຍສະເພາະ, ທ່ານມີຄວາມສົນໃຈໃນຫ້າບໍລິສັດທີ່ມີການຂາຍຮຸ້ນທີ່ໃຫຍ່ທີ່ສຸດໃນແຕ່ລະອຸດສາຫະກໍາ.

ການຮວບຮວມດັ່ງກ່າວຈະຮຽກຮ້ອງໃຫ້ມີຫຼາຍຂັ້ນຕອນຕໍ່ໄປນີ້ເພື່ອແປຂໍ້ມູນເຂົ້າໄປໃນຮູບແບບທີ່ຕ້ອງການ (ເວົ້າໃນຄໍາສັບທົ່ວໄປ).

1. ສ້າງແຫຼ່ງທີ່ອີງໃສ່ຫົວຂໍ້ທີ່ເຜີຍແຜ່ຂໍ້ມູນການຊື້ຂາຍຫຼັກຊັບດິບ. ພວກເຮົາຈະມີແຜນທີ່ວັດຖຸຂອງປະເພດ StockTransaction ກັບວັດຖຸຂອງປະເພດ ShareVolume. ຈຸດແມ່ນວ່າວັດຖຸ StockTransaction ມີ metadata ການຂາຍ, ແຕ່ພວກເຮົາຕ້ອງການພຽງແຕ່ຂໍ້ມູນກ່ຽວກັບຈໍານວນຮຸ້ນທີ່ຖືກຂາຍ.
2. ກຸ່ມຂໍ້ມູນ ShareVolume ໂດຍສັນຍາລັກຫຼັກຊັບ. ເມື່ອຈັດກຸ່ມໂດຍສັນຍາລັກ, ທ່ານສາມາດຫຍໍ້ຂໍ້ມູນນີ້ເຂົ້າໄປໃນຈໍານວນຍ່ອຍຂອງປະລິມານການຂາຍຫຼັກຊັບ. ມັນເປັນມູນຄ່າທີ່ສັງເກດວ່າວິທີການ KStream.groupBy ຕອບຕົວຢ່າງຂອງປະເພດ KGroupedStream. ແລະທ່ານສາມາດໄດ້ຮັບຕົວຢ່າງ KTable ໂດຍການໂທຫາວິທີການເພີ່ມເຕີມ KGroupedStream.reduce.

ການໂຕ້ຕອບ KGroupedStream ແມ່ນຫຍັງ

ວິທີການ KStream.groupBy ແລະ KStream.groupByKey ສົ່ງຄືນຕົວຢ່າງຂອງ KGroupedStream. KGroupedStream ແມ່ນການເປັນຕົວແທນຂັ້ນກາງຂອງກະແສເຫດການຫຼັງຈາກຈັດກຸ່ມໂດຍກະແຈ. ມັນບໍ່ໄດ້ມີຈຸດປະສົງໃນການເຮັດວຽກໂດຍກົງກັບມັນ. ແທນທີ່ຈະ, KGroupedStream ຖືກນໍາໃຊ້ສໍາລັບການດໍາເນີນງານການລວບລວມ, ຜົນໄດ້ຮັບແມ່ນ KTable ສະເຫມີ. ແລະເນື່ອງຈາກຜົນຂອງການດໍາເນີນງານການລວບລວມແມ່ນ KTable ແລະພວກເຂົາໃຊ້ຮ້ານຂອງລັດ, ມັນເປັນໄປໄດ້ວ່າບໍ່ແມ່ນການປັບປຸງທັງຫມົດເປັນຜົນມາຈາກການສົ່ງຕໍ່ລົງໄປໃນທໍ່ນັ້ນ.

ວິທີການ KTable.groupBy ສົ່ງຄືນ KGroupedTable ທີ່ຄ້າຍຄືກັນ - ການສະແດງລະດັບປານກາງຂອງກະແສການອັບເດດ, ຈັດກຸ່ມໃຫມ່ໂດຍລະຫັດ.

ໃຫ້ໃຊ້ເວລາພັກຜ່ອນສັ້ນແລະເບິ່ງຮູບ. 5.9, ເຊິ່ງສະແດງໃຫ້ເຫັນສິ່ງທີ່ພວກເຮົາບັນລຸໄດ້. topology ນີ້ຄວນຈະຄຸ້ນເຄີຍກັບທ່ານຫຼາຍ.

ຕອນນີ້ໃຫ້ເບິ່ງລະຫັດສໍາລັບ topology ນີ້ (ມັນສາມາດພົບໄດ້ໃນໄຟລ໌ src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (ລາຍການ 5.2).

ລະຫັດທີ່ໃຫ້ໄດ້ຖືກຈໍາແນກໂດຍຄວາມຫຍໍ້ຂອງມັນແລະປະລິມານຂະຫນາດໃຫຍ່ຂອງການປະຕິບັດໃນຫຼາຍສາຍ. ທ່ານອາດຈະສັງເກດເຫັນສິ່ງໃຫມ່ໃນພາລາມິເຕີທໍາອິດຂອງວິທີການ builder.stream: ຄ່າຂອງປະເພດ enum AutoOffsetReset.EARLIEST (ຍັງມີ LATEST), ກໍານົດໂດຍໃຊ້ວິທີການ Consumed.withOffsetResetPolicy. ປະເພດການນັບເລກນີ້ສາມາດຖືກໃຊ້ເພື່ອລະບຸຍຸດທະສາດການປັບຄ່າຊົດເຊີຍສຳລັບແຕ່ລະ KStream ຫຼື KTable ແລະມີຄວາມສຳຄັນເໜືອຕົວເລືອກການຕັ້ງຄືນຄ່າຊົດເຊີຍຈາກການກຳນົດຄ່າ.

GroupByKey ແລະ GroupBy

ການໂຕ້ຕອບ KStream ມີສອງວິທີສໍາລັບການຈັດກຸ່ມບັນທຶກ: GroupByKey ແລະ GroupBy. ທັງສອງສົ່ງຄືນ KGroupedTable, ດັ່ງນັ້ນທ່ານອາດຈະສົງໄສວ່າຄວາມແຕກຕ່າງລະຫວ່າງພວກມັນແມ່ນຫຍັງແລະເວລາໃດທີ່ຈະໃຊ້ອັນໃດ?

ວິທີການ GroupByKey ຖືກນໍາໃຊ້ໃນເວລາທີ່ກະແຈໃນ KStream ແມ່ນບໍ່ຫວ່າງເປົ່າ. ແລະສໍາຄັນທີ່ສຸດ, ທຸງ "ຮຽກຮ້ອງໃຫ້ມີການແບ່ງສ່ວນໃຫມ່" ບໍ່ເຄີຍຖືກຕັ້ງ.

ວິທີການ GroupBy ສົມມຸດວ່າທ່ານໄດ້ປ່ຽນລະຫັດການຈັດກຸ່ມ, ດັ່ງນັ້ນທຸງ repartition ຖືກຕັ້ງເປັນຖືກຕ້ອງ. ການປະຕິບັດການເຂົ້າຮ່ວມ, ການລວບລວມ, ແລະອື່ນໆຫຼັງຈາກວິທີການ GroupBy ຈະເຮັດໃຫ້ການແບ່ງສ່ວນໃຫມ່ອັດຕະໂນມັດ.
ສະຫຼຸບ: ເມື່ອໃດກໍ່ຕາມທີ່ເປັນໄປໄດ້, ທ່ານຄວນໃຊ້ GroupByKey ແທນທີ່ຈະ GroupBy.

ມັນເປັນທີ່ຊັດເຈນວ່າ mapValues ​​​​ແລະ groupBymethod ເຮັດແນວໃດ, ສະນັ້ນໃຫ້ພວກເຮົາພິຈາລະນາ sum() method (ພົບໃນ src/main/java/bbejeck/model/ShareVolume.java) (Listing 5.3).

ວິທີການ ShareVolume.sum ສົ່ງຄືນການແລ່ນທັງໝົດຂອງປະລິມານການຂາຍຫຼັກຊັບ, ແລະຜົນຂອງການຄິດໄລ່ທັງໝົດຂອງຕ່ອງໂສ້ການຄິດໄລ່ແມ່ນວັດຖຸ KTable. . ຕອນນີ້ເຈົ້າເຂົ້າໃຈບົດບາດຂອງ KTable ແລ້ວ. ເມື່ອວັດຖຸ ShareVolume ມາຮອດ, ວັດຖຸ KTable ທີ່ສອດຄ້ອງກັນເກັບຮັກສາການປັບປຸງປະຈຸບັນຫລ້າສຸດ. ມັນເປັນສິ່ງສໍາຄັນທີ່ຈະຈື່ຈໍາວ່າການປັບປຸງທັງຫມົດແມ່ນສະທ້ອນໃຫ້ເຫັນໃນ shareVolumeKTable ທີ່ຜ່ານມາ, ແຕ່ບໍ່ແມ່ນທັງຫມົດທີ່ຖືກສົ່ງຕໍ່ໄປ.

ຫຼັງຈາກນັ້ນ, ພວກເຮົາໃຊ້ KTable ນີ້ເພື່ອລວບລວມ (ໂດຍຈໍານວນຮຸ້ນທີ່ຊື້ຂາຍ) ເພື່ອມາຮອດຫ້າບໍລິສັດທີ່ມີປະລິມານການຊື້ຂາຍສູງສຸດໃນແຕ່ລະອຸດສາຫະກໍາ. ການປະຕິບັດຂອງພວກເຮົາໃນກໍລະນີນີ້ຈະຄ້າຍຄືກັບການລວບລວມຄັ້ງທໍາອິດ.

1. ດໍາເນີນການກຸ່ມອື່ນໂດຍການດໍາເນີນງານເພື່ອຈັດກຸ່ມວັດຖຸ ShareVolume ສ່ວນບຸກຄົນໂດຍອຸດສາຫະກໍາ.
2. ເລີ່ມສະຫຼຸບວັດຖຸ ShareVolume. ເວລານີ້ວັດຖຸການລວບລວມແມ່ນແຖວບູລິມະສິດທີ່ມີຂະຫນາດຄົງທີ່. ໃນແຖວທີ່ມີຂະຫນາດຄົງທີ່ນີ້, ພຽງແຕ່ຫ້າບໍລິສັດທີ່ມີຈໍານວນຮຸ້ນທີ່ຂາຍຫຼາຍທີ່ສຸດແມ່ນຖືກຮັກສາໄວ້.
3. ແຜນຜັງແຖວຈາກວັກກ່ອນຫນ້າໄປຫາຄ່າສະຕຣິງແລະສົ່ງຄືນຫ້າຫຼັກຊັບທີ່ມີການຊື້ຂາຍຫຼາຍທີ່ສຸດໂດຍຕົວເລກໂດຍອຸດສາຫະກໍາ.
4. ຂຽນຜົນໄດ້ຮັບໃນຮູບແບບສະຕຣິງໃສ່ຫົວຂໍ້.

ໃນຮູບ. ຮູບທີ 5.10 ສະແດງເສັ້ນສະແດງການໄຫຼເຂົ້າຂອງຂໍ້ມູນ. ດັ່ງທີ່ທ່ານສາມາດເຫັນໄດ້, ຮອບທີສອງຂອງການປຸງແຕ່ງແມ່ນຂ້ອນຂ້າງງ່າຍດາຍ.

ຕອນນີ້ພວກເຮົາມີຄວາມເຂົ້າໃຈຢ່າງຈະແຈ້ງກ່ຽວກັບໂຄງສ້າງຂອງການປະມວນຜົນຮອບທີສອງນີ້, ພວກເຮົາສາມາດຫັນໄປຫາລະຫັດແຫຼ່ງຂອງມັນ (ທ່ານຈະເຫັນມັນຢູ່ໃນໄຟລ໌ src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java) (ລາຍຊື່ 5.4) .

ຕົວແປເບື້ອງຕົ້ນນີ້ມີຕົວແປຄົງທີ່Queue. ນີ້ແມ່ນວັດຖຸທີ່ກຳນົດເອງທີ່ເປັນອະແດັບເຕີສຳລັບ java.util.TreeSet ທີ່ໃຊ້ເພື່ອຕິດຕາມຜົນ N ເທິງສຸດໃນການຈັດຮຽງລຳດັບຂອງຮຸ້ນທີ່ຊື້ຂາຍກັນ.

ທ່ານໄດ້ເຫັນການໂທ groupBy ແລະ mapValues ​​ແລ້ວ, ດັ່ງນັ້ນພວກເຮົາຈະບໍ່ເຂົ້າໄປໃນການເອີ້ນເຫຼົ່ານັ້ນ (ພວກເຮົາກໍາລັງໂທຫາວິທີການ KTable.toStream ເພາະວ່າວິທີການ KTable.print ຖືກຍົກເລີກ). ແຕ່ທ່ານຍັງບໍ່ໄດ້ເຫັນສະບັບ KTable ຂອງ aggregate() ເທື່ອ, ດັ່ງນັ້ນພວກເຮົາຈະໃຊ້ເວລາເລັກນ້ອຍເພື່ອປຶກສາຫາລືກັນ.

ຕາມທີ່ທ່ານຈື່, ສິ່ງທີ່ເຮັດໃຫ້ KTable ແຕກຕ່າງກັນແມ່ນວ່າບັນທຶກທີ່ມີລະຫັດດຽວກັນຖືກພິຈາລະນາການປັບປຸງ. KTable ແທນລາຍການເກົ່າດ້ວຍອັນໃໝ່. ການຮວບຮວມເກີດຂຶ້ນໃນລັກສະນະທີ່ຄ້າຍຄືກັນ: ບັນທຶກຫຼ້າສຸດທີ່ມີລະຫັດດຽວກັນຖືກລວບລວມ. ເມື່ອບັນທຶກມາຮອດ, ມັນຖືກເພີ່ມເຂົ້າໃນຊັ້ນ FixedSizePriorityQueue ໂດຍໃຊ້ adder (ພາລາມິເຕີທີສອງໃນການໂທວິທີການລວມ), ແຕ່ຖ້າມີບັນທຶກອື່ນທີ່ມີລະຫັດດຽວກັນ, ຫຼັງຈາກນັ້ນບັນທຶກເກົ່າຖືກໂຍກຍ້າຍອອກໂດຍໃຊ້ຕົວຫັກ (ພາລາມິເຕີທີສາມໃນ ໂທ​ຫາ​ວິ​ທີ​ການ​ລວມ​)​.

ນີ້ທັງຫມົດຫມາຍຄວາມວ່າການລວບລວມຂອງພວກເຮົາ, FixedSizePriorityQueue, ບໍ່ໄດ້ລວມມູນຄ່າທັງຫມົດດ້ວຍກຸນແຈຫນຶ່ງ, ແຕ່ເກັບຮັກສາຈໍານວນການເຄື່ອນຍ້າຍຂອງປະລິມານຂອງ N ປະເພດການຊື້ຂາຍຫຼາຍທີ່ສຸດຂອງຫຼັກຊັບ. ແຕ່ລະລາຍການທີ່ເຂົ້າມາມີຈໍານວນຮຸ້ນທັງໝົດທີ່ຂາຍມາເຖິງຕອນນັ້ນ. KTable ຈະໃຫ້ທ່ານຂໍ້ມູນກ່ຽວກັບວ່າຮຸ້ນຂອງບໍລິສັດໃດທີ່ມີການຊື້ຂາຍຫຼາຍທີ່ສຸດ, ໂດຍບໍ່ມີການຮຽກຮ້ອງໃຫ້ມີການລວບລວມການປັບປຸງແຕ່ລະຄັ້ງ.

ພວກເຮົາໄດ້ຮຽນຮູ້ທີ່ຈະເຮັດສອງສິ່ງທີ່ສໍາຄັນ:

• ກຸ່ມຄ່າໃນ KTable ໂດຍຄີທົ່ວໄປ;
• ປະຕິບັດການປະຕິບັດທີ່ເປັນປະໂຫຍດເຊັ່ນ: ການລວບລວມແລະການລວບລວມຢູ່ໃນຄ່າກຸ່ມເຫຼົ່ານີ້.

ການຮູ້ວິທີການປະຕິບັດການເຫຼົ່ານີ້ແມ່ນສໍາຄັນຕໍ່ການເຂົ້າໃຈຄວາມຫມາຍຂອງຂໍ້ມູນທີ່ເຄື່ອນຍ້າຍຜ່ານຄໍາຮ້ອງສະຫມັກ Kafka Streams ແລະເຂົ້າໃຈສິ່ງທີ່ມັນປະຕິບັດ.

ພວກ​ເຮົາ​ຍັງ​ໄດ້​ນໍາ​ເອົາ​ບາງ​ແນວ​ຄວາມ​ຄິດ​ທີ່​ສໍາ​ຄັນ​ທີ່​ໄດ້​ປຶກ​ສາ​ຫາ​ລື​ກ່ອນ​ຫນ້າ​ນີ້​ໃນ​ປຶ້ມ​ນີ້​. ໃນບົດທີ 4, ພວກເຮົາໄດ້ປຶກສາຫາລືກ່ຽວກັບຄວາມທົນທານຕໍ່ຄວາມຜິດ, ລັດທ້ອງຖິ່ນມີຄວາມສໍາຄັນແນວໃດສໍາລັບຄໍາຮ້ອງສະຫມັກການຖ່າຍທອດ. ຕົວຢ່າງທໍາອິດໃນບົດນີ້ສະແດງໃຫ້ເຫັນວ່າເປັນຫຍັງລັດທ້ອງຖິ່ນຈຶ່ງມີຄວາມສໍາຄັນຫຼາຍ - ມັນຊ່ວຍໃຫ້ທ່ານສາມາດຕິດຕາມຂໍ້ມູນທີ່ທ່ານໄດ້ເຫັນແລ້ວ. ການເຂົ້າເຖິງທ້ອງຖິ່ນຫຼີກເວັ້ນຄວາມລ່າຊ້າຂອງເຄືອຂ່າຍ, ເຮັດໃຫ້ແອັບພລິເຄຊັນມີປະສິດທິພາບຫຼາຍຂຶ້ນແລະທົນທານຕໍ່ຄວາມຜິດພາດ.

ເມື່ອປະຕິບັດການລວບລວມຫຼືການລວບລວມໃດໆ, ທ່ານຕ້ອງລະບຸຊື່ຂອງຮ້ານຂອງລັດ. ການດໍາເນີນງານການລວບລວມແລະການລວບລວມຄືນຕົວຢ່າງ KTable, ແລະ KTable ໃຊ້ບ່ອນເກັບຂໍ້ມູນຂອງລັດເພື່ອທົດແທນຜົນໄດ້ຮັບເກົ່າດ້ວຍອັນໃຫມ່. ດັ່ງທີ່ທ່ານໄດ້ເຫັນ, ບໍ່ແມ່ນການອັບເດດທັງຫມົດຖືກສົ່ງລົງໄປໃນທໍ່, ແລະນີ້ເປັນສິ່ງສໍາຄັນເພາະວ່າການດໍາເນີນງານການລວບລວມໄດ້ຖືກອອກແບບເພື່ອຜະລິດຂໍ້ມູນສະຫຼຸບ. ຖ້າ​ຫາກ​ວ່າ​ທ່ານ​ບໍ່​ໄດ້​ນໍາ​ໃຊ້​ລັດ​ທ້ອງ​ຖິ່ນ​, KTable ຈະ​ສົ່ງ​ຕໍ່​ຜົນ​ການ​ລວມ​ທັງ​ຫມົດ​ແລະ​ການ​ມ້ວນ​.

ຕໍ່ໄປ, ພວກເຮົາຈະເບິ່ງການປະຕິບັດງານເຊັ່ນການລວບລວມພາຍໃນໄລຍະເວລາສະເພາະໃດຫນຶ່ງ - ອັນທີ່ເອີ້ນວ່າການດໍາເນີນງານ windowing.

5.3.2. ການດໍາເນີນງານຂອງປ່ອງຢ້ຽມ

ໃນພາກກ່ອນ, ພວກເຮົາໄດ້ນໍາສະເຫນີການເລື່ອນແລະການລວບລວມ. ຄໍາຮ້ອງສະຫມັກດັ່ງກ່າວໄດ້ປະຕິບັດການຂາຍຫຼັກຊັບຢ່າງຕໍ່ເນື່ອງຕໍ່ເນື່ອງໂດຍການລວບລວມຫ້າຫຼັກຊັບທີ່ມີການຊື້ຂາຍຫຼາຍທີ່ສຸດໃນການແລກປ່ຽນ.

ບາງຄັ້ງການຮວບຮວມຢ່າງຕໍ່ເນື່ອງແລະການມ້ວນຜົນໄດ້ຮັບແມ່ນມີຄວາມຈໍາເປັນ. ແລະບາງຄັ້ງທ່ານຈໍາເປັນຕ້ອງປະຕິບັດການດໍາເນີນງານພຽງແຕ່ໃນໄລຍະເວລາໃດຫນຶ່ງ. ຕົວຢ່າງ, ຄິດໄລ່ຈໍານວນການຊື້ຂາຍແລກປ່ຽນກັບຮຸ້ນຂອງບໍລິສັດສະເພາະໃນ 10 ນາທີທີ່ຜ່ານມາ. ຫຼືຈໍານວນຜູ້ໃຊ້ຄລິກໃສ່ປ້າຍໂຄສະນາໃຫມ່ໃນ 15 ນາທີທີ່ຜ່ານມາ. ແອັບພລິເຄຊັນອາດຈະດໍາເນີນການດັ່ງກ່າວຫຼາຍຄັ້ງ, ແຕ່ຜົນໄດ້ຮັບທີ່ນໍາໃຊ້ກັບໄລຍະເວລາທີ່ກໍານົດໄວ້ເທົ່ານັ້ນ (ຫນ້າຕ່າງເວລາ).

ການນັບການຊື້ຂາຍແລກປ່ຽນໂດຍຜູ້ຊື້

ໃນຕົວຢ່າງຕໍ່ໄປ, ພວກເຮົາຈະຕິດຕາມທຸລະກໍາຫຼັກຊັບໃນທົ່ວພໍ່ຄ້າຫຼາຍ - ອົງການຈັດຕັ້ງຂະຫນາດໃຫຍ່ຫຼືນັກການເງິນສ່ວນບຸກຄົນທີ່ສະຫລາດ.

ມີສອງເຫດຜົນທີ່ເປັນໄປໄດ້ສໍາລັບການຕິດຕາມນີ້. ຫນຶ່ງໃນນັ້ນແມ່ນຄວາມຕ້ອງການທີ່ຈະຮູ້ວ່າຜູ້ນໍາຕະຫຼາດຊື້ / ຂາຍແມ່ນຫຍັງ. ຖ້າຜູ້ຫຼິ້ນໃຫຍ່ເຫຼົ່ານີ້ແລະນັກລົງທຶນທີ່ມີຄວາມຊັບຊ້ອນເຫັນໂອກາດ, ມັນເຮັດໃຫ້ຄວາມຮູ້ສຶກທີ່ຈະປະຕິບັດຕາມຍຸດທະສາດຂອງພວກເຂົາ. ເຫດຜົນທີສອງແມ່ນຄວາມປາຖະຫນາທີ່ຈະເຫັນສັນຍານທີ່ເປັນໄປໄດ້ຂອງການຊື້ຂາຍພາຍໃນທີ່ຜິດກົດຫມາຍ. ເພື່ອເຮັດສິ່ງນີ້, ທ່ານຈະຕ້ອງວິເຄາະຄວາມສໍາພັນຂອງຍອດຂາຍຂະຫນາດໃຫຍ່ທີ່ມີການປ່ອຍຂ່າວທີ່ສໍາຄັນ.

ການຕິດຕາມດັ່ງກ່າວປະກອບດ້ວຍຂັ້ນຕອນຕໍ່ໄປນີ້:

• ການສ້າງກະແສສໍາລັບການອ່ານຈາກຫົວຂໍ້ທຸລະກໍາຫຼັກຊັບ;
• ຈັດກຸ່ມບັນທຶກຂາເຂົ້າໂດຍ ID ຜູ້ຊື້ ແລະສັນຍາລັກຫຼັກຊັບ. ການໂທຫາກຸ່ມໂດຍວິທີການສົ່ງຄືນຕົວຢ່າງຂອງຊັ້ນ KGroupedStream;
• ວິທີການ KGroupedStream.windowedBy ຕອບສະຕຣີມຂໍ້ມູນທີ່ຈຳກັດຢູ່ໃນໜ້າຕ່າງເວລາ, ເຊິ່ງອະນຸຍາດໃຫ້ມີການລວມກັນແບບ windowed. ອີງຕາມປະເພດຂອງປ່ອງຢ້ຽມ, ບໍ່ວ່າຈະເປັນ TimeWindowedKStream ຫຼື SessionWindowedKStream ຈະຖືກສົ່ງຄືນ;
• ຈໍານວນການເຮັດທຸລະກໍາສໍາລັບການດໍາເນີນງານການລວບລວມ. ການໄຫຼເຂົ້າຂອງຂໍ້ມູນ windowed ກໍານົດວ່າບັນທຶກສະເພາະໃດຫນຶ່ງໄດ້ຖືກພິຈາລະນາໃນການນັບນີ້;
• ການຂຽນຜົນໄດ້ຮັບໄປຫາຫົວຂໍ້ໃດຫນຶ່ງຫຼືສົ່ງໃຫ້ພວກເຂົາຢູ່ໃນ console ໃນລະຫວ່າງການພັດທະນາ.

topology ຂອງຄໍາຮ້ອງສະຫມັກນີ້ແມ່ນງ່າຍດາຍ, ແຕ່ຮູບພາບທີ່ຊັດເຈນຂອງມັນຈະເປັນປະໂຫຍດ. ລອງເບິ່ງຮູບ. 5.11.

ຕໍ່ໄປ, ພວກເຮົາຈະເບິ່ງການທໍາງານຂອງການດໍາເນີນງານຂອງປ່ອງຢ້ຽມແລະລະຫັດທີ່ສອດຄ້ອງກັນ.

ປະເພດປ່ອງຢ້ຽມ

ມີສາມປະເພດຂອງປ່ອງຢ້ຽມໃນ Kafka Stream:

• ກອງປະຊຸມ;
• “tumbling” (tumbling);
• ເລື່ອນ / hopping.

ອັນໃດທີ່ຈະເລືອກແມ່ນຂຶ້ນກັບຄວາມຕ້ອງການຂອງທຸລະກິດຂອງທ່ານ. ໜ້າຈໍເລື່ອນ ແລະ ໂດດແມ່ນຈຳກັດເວລາ, ໃນຂະນະທີ່ໜ້າຕ່າງຂອງເຊດຊັນຖືກຈຳກັດໂດຍການເຄື່ອນໄຫວຂອງຜູ້ໃຊ້ - ໄລຍະເວລາຂອງເຊດຊັນແມ່ນກຳນົດຢ່າງດຽວໂດຍວິທີທີ່ຜູ້ໃຊ້ມີການເຄື່ອນໄຫວ. ສິ່ງທີ່ສໍາຄັນທີ່ຕ້ອງຈື່ແມ່ນວ່າທຸກປະເພດປ່ອງຢ້ຽມແມ່ນອີງໃສ່ການປະທັບຕາວັນທີ / ເວລາຂອງລາຍການ, ບໍ່ແມ່ນເວລາຂອງລະບົບ.

ຕໍ່ໄປ, ພວກເຮົາປະຕິບັດ topology ຂອງພວກເຮົາກັບແຕ່ລະປະເພດປ່ອງຢ້ຽມ. ລະຫັດຄົບຖ້ວນຈະຖືກມອບໃຫ້ພຽງແຕ່ໃນຕົວຢ່າງທໍາອິດ;

ປ່ອງຢ້ຽມຂອງເຊດຊັນ

ປ່ອງຢ້ຽມ Session ແມ່ນແຕກຕ່າງກັນຫຼາຍຈາກປ່ອງຢ້ຽມປະເພດອື່ນໆ. ພວກເຂົາເຈົ້າໄດ້ຖືກຈໍາກັດບໍ່ຫຼາຍປານໃດໂດຍທີ່ໃຊ້ເວລາເປັນກິດຈະກໍາຂອງຜູ້ໃຊ້ (ຫຼືກິດຈະກໍາຂອງຫນ່ວຍງານທີ່ທ່ານຕ້ອງການທີ່ຈະຕິດຕາມ). ໜ້າຈໍເຊດຊັນຖືກຈຳກັດໂດຍໄລຍະເວລາທີ່ບໍ່ມີການເຄື່ອນໄຫວ.

ຮູບທີ 5.12 ສະແດງແນວຄວາມຄິດຂອງໜ້າຕ່າງກອງປະຊຸມ. ເຊດຊັນທີ່ນ້ອຍກວ່າຈະຮວມກັບເຊດຊັນທາງຊ້າຍຂອງມັນ. ແລະກອງປະຊຸມທາງດ້ານຂວາຈະແຍກກັນເພາະວ່າມັນປະຕິບັດຕາມໄລຍະເວລາທີ່ຍາວນານທີ່ບໍ່ມີການເຄື່ອນໄຫວ. ປ່ອງຢ້ຽມຂອງເຊດຊັນແມ່ນອີງໃສ່ການເຄື່ອນໄຫວຂອງຜູ້ໃຊ້, ແຕ່ໃຊ້ການປະທັບຕາວັນທີ/ເວລາຈາກລາຍການເພື່ອກຳນົດວ່າເຊສຊັນໃດທີ່ເຂົ້າເປັນຂອງ.

ການນໍາໃຊ້ປ່ອງຢ້ຽມຂອງກອງປະຊຸມເພື່ອຕິດຕາມທຸລະກໍາຫຼັກຊັບ

ໃຫ້ໃຊ້ປ່ອງຢ້ຽມຂອງເຊດຊັນເພື່ອເກັບກໍາຂໍ້ມູນກ່ຽວກັບການເຮັດທຸລະກໍາແລກປ່ຽນ. ການປະຕິບັດຫນ້າຕ່າງຂອງກອງປະຊຸມແມ່ນສະແດງຢູ່ໃນລາຍການ 5.5 (ເຊິ່ງສາມາດພົບໄດ້ໃນ src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java).

ທ່ານໄດ້ເຫັນການດໍາເນີນການສ່ວນໃຫຍ່ໃນ topology ນີ້ແລ້ວ, ດັ່ງນັ້ນບໍ່ຈໍາເປັນຕ້ອງເບິ່ງພວກມັນອີກເທື່ອຫນຶ່ງຢູ່ທີ່ນີ້. ແຕ່ຍັງມີຫຼາຍອົງປະກອບໃຫມ່ຢູ່ທີ່ນີ້, ທີ່ພວກເຮົາຈະປຶກສາຫາລືໃນປັດຈຸບັນ.

ການດໍາເນີນງານຂອງກຸ່ມໂດຍປົກກະຕິຈະປະຕິບັດບາງປະເພດຂອງການລວບລວມ (ການລວບລວມ, ການລວບລວມ, ຫຼືການນັບ). ທ່ານ​ສາ​ມາດ​ປະ​ຕິ​ບັດ​ທັງ​ການ​ສະ​ສົມ​ສະ​ສົມ​ທີ່​ມີ​ທັງ​ຫມົດ​ທີ່​ແລ່ນ​, ຫຼື​ການ​ລວບ​ລວມ​ປ່ອງ​ຢ້ຽມ​, ເຊິ່ງ​ໃຊ້​ເວ​ລາ​ເຂົ້າ​ໄປ​ໃນ​ບັນ​ຊີ​ການ​ບັນ​ທຶກ​ພາຍ​ໃນ​ປ່ອງ​ຢ້ຽມ​ທີ່​ໃຊ້​ເວ​ລາ​ທີ່​ກໍາ​ນົດ​ໄວ້​.

ລະຫັດໃນລາຍການ 5.5 ຈະນັບຈໍານວນການເຮັດທຸລະກໍາພາຍໃນປ່ອງຢ້ຽມຂອງເຊດຊັນ. ໃນຮູບ. 5.13 ການກະທຳເຫຼົ່ານີ້ຖືກວິເຄາະເທື່ອລະຂັ້ນ.

ໂດຍການໂທຫາ windowedBy(SessionWindows.with(twentySeconds).ຈົນກ່ວາ(fifteenMinutes)) ພວກເຮົາສ້າງໜ້າຕ່າງເຊດຊັນທີ່ມີຊ່ວງເວລາບໍ່ມີການເຄື່ອນໄຫວຂອງ 20 ວິນາທີ ແລະໄລຍະຕໍ່ເນື່ອງຂອງ 15 ນາທີ. ໄລຍະຫ່າງທີ່ບໍ່ເຮັດວຽກຂອງ 20 ວິນາທີຫມາຍຄວາມວ່າແອັບພລິເຄຊັນຈະລວມເອົາລາຍການທີ່ເຂົ້າມາພາຍໃນ 20 ວິນາທີຂອງການສິ້ນສຸດຫຼືການເລີ່ມຕົ້ນຂອງເຊດຊັນໃນປະຈຸບັນເຂົ້າໄປໃນກອງປະຊຸມ (ການເຄື່ອນໄຫວ) ໃນປັດຈຸບັນ.

ຕໍ່ໄປ, ພວກເຮົາລະບຸວ່າການປະຕິບັດການລວບລວມທີ່ຕ້ອງໄດ້ຮັບການປະຕິບັດຢູ່ໃນປ່ອງຢ້ຽມຂອງກອງປະຊຸມ - ໃນກໍລະນີນີ້, ນັບ. ຖ້າລາຍການຂາເຂົ້າຕົກຢູ່ນອກປ່ອງຢ້ຽມທີ່ບໍ່ມີການເຄື່ອນໄຫວ (ທັງສອງດ້ານຂອງສະແຕມວັນທີ / ເວລາ), ແອັບພລິເຄຊັນຈະສ້າງເຊດຊັນໃຫມ່. ໄລຍະການເກັບຮັກສາໄວ້ຫມາຍເຖິງການຮັກສາເຊດຊັນສໍາລັບຈໍານວນເວລາທີ່ແນ່ນອນແລະອະນຸຍາດໃຫ້ມີຂໍ້ມູນຊັກຊ້າທີ່ຂະຫຍາຍອອກໄປເກີນໄລຍະເວລາທີ່ບໍ່ມີການເຄື່ອນໄຫວຂອງເຊດຊັນແຕ່ຍັງສາມາດຕິດໄດ້. ນອກຈາກນັ້ນ, ການເລີ່ມຕົ້ນແລະສິ້ນສຸດຂອງກອງປະຊຸມໃຫມ່ທີ່ເກີດຈາກການລວມກັນແມ່ນກົງກັບວັນທີ / ເວລາທີ່ໄວທີ່ສຸດແລະຫລ້າສຸດ.

ຂໍໃຫ້ເບິ່ງບາງລາຍການຈາກວິທີການນັບເພື່ອເບິ່ງວ່າເຊດຊັນເຮັດວຽກແນວໃດ (ຕາຕະລາງ 5.1).

ເມື່ອບັນທຶກມາຮອດ, ພວກເຮົາຊອກຫາເຊດຊັນທີ່ມີຢູ່ແລ້ວດ້ວຍກະແຈດຽວກັນ, ເວລາສິ້ນສຸດໜ້ອຍກວ່າສະແຕມວັນທີ/ເວລາປັດຈຸບັນ - ຊ່ວງເວລາບໍ່ມີການເຄື່ອນໄຫວ, ແລະເວລາເລີ່ມຕົ້ນທີ່ໃຫຍ່ກວ່າສະແຕມວັນທີ/ເວລາປັດຈຸບັນ + ຊ່ວງເວລາທີ່ບໍ່ມີການເຄື່ອນໄຫວ. ການພິຈາລະນານີ້ເຂົ້າໄປໃນບັນຊີ, ສີ່ລາຍການຈາກຕາຕະລາງ. 5.1 ຖືກລວມເຂົ້າເປັນກອງປະຊຸມດຽວດັ່ງຕໍ່ໄປນີ້.

1. ບັນທຶກ 1 ມາຮອດກ່ອນ, ດັ່ງນັ້ນເວລາເລີ່ມຕົ້ນແມ່ນເທົ່າກັບເວລາສິ້ນສຸດແລະເປັນ 00:00:00.

2. ຕໍ່ໄປ, ເຂົ້າ 2 ມາຮອດ, ແລະພວກເຮົາຊອກຫາກອງປະຊຸມທີ່ສິ້ນສຸດບໍ່ໄວກວ່າ 23:59:55 ແລະເລີ່ມຕົ້ນບໍ່ເກີນ 00:00:35. ພວກເຮົາຊອກຫາບັນທຶກ 1 ແລະປະສົມປະສານຂອງກອງປະຊຸມ 1 ແລະ 2. ພວກເຮົາໃຊ້ເວລາເລີ່ມຕົ້ນຂອງກອງປະຊຸມ 1 (ກ່ອນຫນ້າ) ແລະເວລາສິ້ນສຸດຂອງກອງປະຊຸມ 2 (ຕໍ່ມາ), ດັ່ງນັ້ນກອງປະຊຸມໃຫມ່ຂອງພວກເຮົາເລີ່ມຕົ້ນໃນເວລາ 00:00:00 ແລະສິ້ນສຸດໃນເວລາ 00: 00:15.

3. ບັນທຶກ 3 ມາຮອດ, ພວກເຮົາຊອກຫາກອງປະຊຸມລະຫວ່າງ 00:00:30 ແລະ 00:01:10 ແລະບໍ່ພົບໃດໆ. ເພີ່ມກອງປະຊຸມທີສອງສໍາລັບທີ່ສໍາຄັນ 123-345-654,FFBE, ເລີ່ມແລະສິ້ນສຸດໃນເວລາ 00:00:50.

4. ບັນທຶກ 4 ມາຮອດແລະພວກເຮົາກໍາລັງຊອກຫາກອງປະຊຸມລະຫວ່າງ 23:59:45 ແລະ 00:00:25. ເວລານີ້ທັງສອງພາກທີ 1 ແລະ 2 ແມ່ນພົບທັງສາມພາກທີ່ລວມເຂົ້າກັນເປັນຫນຶ່ງ, ມີເວລາເລີ່ມຕົ້ນຂອງ 00:00:00 ແລະເວລາສິ້ນສຸດຂອງ 00:00:15.

ຈາກສິ່ງທີ່ໄດ້ອະທິບາຍໃນພາກນີ້, ມັນເປັນມູນຄ່າທີ່ຈະຈື່ຈໍາ nuances ທີ່ສໍາຄັນດັ່ງຕໍ່ໄປນີ້:

• ເຊດຊັນບໍ່ແມ່ນໜ້າຈໍຂະໜາດຄົງທີ່. ໄລຍະເວລາຂອງກອງປະຊຸມແມ່ນຖືກກໍານົດໂດຍກິດຈະກໍາພາຍໃນໄລຍະເວລາໃດຫນຶ່ງ;
• ການສະແຕມວັນທີ/ເວລາໃນຂໍ້ມູນກຳນົດວ່າເຫດການດັ່ງກ່າວຕົກຢູ່ໃນເຊດຊັນທີ່ມີຢູ່ແລ້ວ ຫຼືໃນລະຫວ່າງໄລຍະເວລາທີ່ບໍ່ເຮັດວຽກ.

ຕໍ່ໄປພວກເຮົາຈະປຶກສາຫາລືປະເພດຂອງປ່ອງຢ້ຽມຕໍ່ໄປ - ປ່ອງຢ້ຽມ "tumbling".

ປ່ອງຢ້ຽມ "Tumbling".

ປ່ອງຢ້ຽມທີ່ລົ້ມລຸກບັນທຶກເຫດການທີ່ຕົກຢູ່ໃນໄລຍະເວລາທີ່ແນ່ນອນ. ຈິນຕະນາການວ່າທ່ານຈໍາເປັນຕ້ອງເກັບກໍາທຸລະກໍາຫຼັກຊັບທັງຫມົດຂອງບໍລິສັດສະເພາະໃດຫນຶ່ງທຸກໆ 20 ວິນາທີ, ດັ່ງນັ້ນທ່ານເກັບກໍາເຫດການທັງຫມົດໃນໄລຍະເວລານັ້ນ. ໃນຕອນທ້າຍຂອງໄລຍະຫ່າງ 20 ວິນາທີ, ປ່ອງຢ້ຽມມ້ວນແລະຍ້າຍໄປໄລຍະການສັງເກດໃຫມ່ 20 ວິນາທີ. ຮູບ 5.14 ສະແດງໃຫ້ເຫັນສະຖານະການນີ້.

ດັ່ງທີ່ທ່ານສາມາດເຫັນໄດ້, ເຫດການທັງຫມົດທີ່ໄດ້ຮັບໃນ 20 ວິນາທີສຸດທ້າຍແມ່ນລວມຢູ່ໃນປ່ອງຢ້ຽມ. ໃນຕອນທ້າຍຂອງໄລຍະເວລານີ້, ປ່ອງຢ້ຽມໃຫມ່ຖືກສ້າງຂື້ນ.

ລາຍຊື່ 5.6 ສະແດງໃຫ້ເຫັນລະຫັດທີ່ສະແດງໃຫ້ເຫັນເຖິງການໃຊ້ windows tumbling ເພື່ອບັນທຶກທຸລະກໍາຫຼັກຊັບທຸກໆ 20 ວິນາທີ (ພົບໃນ src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

ດ້ວຍການປ່ຽນແປງນ້ອຍໆນີ້ຕໍ່ກັບການໂທດ້ວຍວິທີ TimeWindows.of, ທ່ານສາມາດໃຊ້ໜ້າຕ່າງ tumbling. ຕົວຢ່າງນີ້ບໍ່ໄດ້ໂທຫາວິທີການຈົນກ່ວາ (), ດັ່ງນັ້ນໄລຍະການເກັບຮັກສາໄວ້ໃນຕອນຕົ້ນຂອງ 24 ຊົ່ວໂມງຈະຖືກໃຊ້.

ສຸດທ້າຍ, ມັນແມ່ນເວລາທີ່ຈະຍ້າຍໄປສຸດທ້າຍຂອງຕົວເລືອກປ່ອງຢ້ຽມ - "hopping" windows .

ເລື່ອນ ("ໂດດ") ປ່ອງຢ້ຽມ

ປ່ອງຢ້ຽມເລື່ອນ / ເລື່ອນແມ່ນຄ້າຍຄືກັບປ່ອງຢ້ຽມທີ່ລົ້ມລົງ, ແຕ່ມີຄວາມແຕກຕ່າງກັນເລັກນ້ອຍ. ປ່ອງຢ້ຽມເລື່ອນບໍ່ໄດ້ລໍຖ້າຈົນກ່ວາໃນຕອນທ້າຍຂອງໄລຍະເວລາກ່ອນທີ່ຈະສ້າງປ່ອງຢ້ຽມໃຫມ່ເພື່ອປະມວນຜົນເຫດການທີ່ຜ່ານມາ. ພວກເຂົາເລີ່ມຕົ້ນການຄິດໄລ່ໃຫມ່ຫຼັງຈາກໄລຍະເວລາລໍຖ້າຫນ້ອຍກວ່າໄລຍະເວລາປ່ອງຢ້ຽມ.

ເພື່ອສະແດງໃຫ້ເຫັນເຖິງຄວາມແຕກຕ່າງລະຫວ່າງກະໂປງແລະການກະໂດດປ່ອງຢ້ຽມ, ໃຫ້ກັບຄືນໄປຫາຕົວຢ່າງຂອງການນັບການຊື້ຂາຍແລກປ່ຽນຫຼັກຊັບ. ເປົ້າໝາຍຂອງພວກເຮົາແມ່ນຍັງຈະນັບຈໍານວນການເຮັດທຸລະກໍາ, ແຕ່ພວກເຮົາບໍ່ຕ້ອງການລໍຖ້າເວລາທັງໝົດກ່ອນທີ່ຈະອັບເດດເຄົາເຕີ. ແທນທີ່ຈະ, ພວກເຮົາຈະປັບປຸງເຄົາເຕີໃນໄລຍະທີ່ສັ້ນກວ່າ. ຕົວຢ່າງ, ພວກເຮົາຍັງຈະນັບຈໍານວນການເຮັດທຸລະກໍາໃນທຸກໆ 20 ວິນາທີ, ແຕ່ປັບປຸງເຄົາເຕີທຸກໆ 5 ວິນາທີ, ດັ່ງທີ່ສະແດງຢູ່ໃນຮູບ. 5.15. ໃນກໍລະນີນີ້, ພວກເຮົາສິ້ນສຸດດ້ວຍສາມປ່ອງຢ້ຽມຜົນໄດ້ຮັບທີ່ມີຂໍ້ມູນທັບຊ້ອນກັນ.

ລາຍຊື່ 5.7 ສະແດງລະຫັດສໍາລັບການກໍານົດປ່ອງຢ້ຽມເລື່ອນ (ພົບໃນ src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java).

ປ່ອງ​ຢ້ຽມ​ທີ່​ກະ​ຕຸ້ນ​ສາ​ມາດ​ໄດ້​ຮັບ​ການ​ປ່ຽນ​ເປັນ​ປ່ອງ​ຢ້ຽມ hopping ໂດຍ​ການ​ເພີ່ມ​ການ​ໂທ​ກັບ​ວິ​ທີ​ການ advanceBy()​. ໃນຕົວຢ່າງທີ່ສະແດງ, ໄລຍະປະຫຍັດແມ່ນ 15 ນາທີ.

ທ່ານໄດ້ເຫັນໃນພາກນີ້ວິທີການຈໍາກັດຜົນໄດ້ຮັບການລວບລວມກັບປ່ອງຢ້ຽມເວລາ. ໂດຍສະເພາະ, ຂ້າພະເຈົ້າຕ້ອງການໃຫ້ທ່ານຈື່ສາມສິ່ງຕໍ່ໄປນີ້ຈາກພາກນີ້:

• ຂະ​ຫນາດ​ຂອງ​ປ່ອງ​ຢ້ຽມ​ຂອງ​ກອງ​ປະ​ຊຸມ​ແມ່ນ​ບໍ່​ຈໍາ​ກັດ​ໂດຍ​ໄລ​ຍະ​ເວ​ລາ​, ແຕ່​ໂດຍ​ກິດ​ຈະ​ກໍາ​ຂອງ​ຜູ້​ໃຊ້​;
• ປ່ອງຢ້ຽມ "tumbling" ໃຫ້ພາບລວມຂອງເຫດການພາຍໃນໄລຍະເວລາໃດຫນຶ່ງ;
• ໄລຍະເວລາຂອງການໂດດປ່ອງຢ້ຽມຖືກແກ້ໄຂ, ແຕ່ພວກມັນຖືກປັບປຸງເລື້ອຍໆແລະອາດມີລາຍການທັບຊ້ອນກັນໃນທຸກປ່ອງຢ້ຽມ.

ຕໍ່ໄປ, ພວກເຮົາຈະຮຽນຮູ້ວິທີການປ່ຽນ KTable ກັບຄືນໄປບ່ອນ KStream ສໍາລັບການເຊື່ອມຕໍ່.

5.3.3. ກຳລັງເຊື່ອມຕໍ່ວັດຖຸ KStream ແລະ KTable

ໃນບົດທີ 4, ພວກເຮົາໄດ້ປຶກສາຫາລືກ່ຽວກັບການເຊື່ອມຕໍ່ສອງວັດຖຸ KStream. ຕອນນີ້ພວກເຮົາຕ້ອງຮຽນຮູ້ວິທີເຊື່ອມຕໍ່ KTable ແລະ KStream. ນີ້ອາດຈະຈໍາເປັນສໍາລັບເຫດຜົນງ່າຍໆຕໍ່ໄປນີ້. KStream ແມ່ນສະຕຣີມບັນທຶກ, ແລະ KTable ແມ່ນກະແສຂອງການບັນທຶກການອັບເດດ, ແຕ່ບາງຄັ້ງທ່ານອາດຈະຕ້ອງການເພີ່ມບໍລິບົດເພີ່ມເຕີມໃຫ້ກັບສະຕຣີມບັນທຶກໂດຍໃຊ້ການອັບເດດຈາກ KTable.

ໃຫ້ເອົາຂໍ້ມູນຈໍານວນການເຮັດທຸລະກໍາຂອງການແລກປ່ຽນຫຼັກຊັບແລະສົມທົບກັບຂ່າວການແລກປ່ຽນຫຼັກຊັບສໍາລັບອຸດສາຫະກໍາທີ່ກ່ຽວຂ້ອງ. ນີ້ແມ່ນສິ່ງທີ່ທ່ານຕ້ອງເຮັດເພື່ອບັນລຸລະຫັດທີ່ເຈົ້າມີຢູ່ແລ້ວ.

1. ປ່ຽນວັດຖຸ KTable ທີ່ມີຂໍ້ມູນຈໍານວນການເຮັດທຸລະກໍາຫຼັກຊັບເຂົ້າໄປໃນ KStream, ປະຕິບັດຕາມໂດຍການປ່ຽນລະຫັດທີ່ມີລະຫັດທີ່ຊີ້ບອກຂະແຫນງອຸດສາຫະກໍາທີ່ສອດຄ້ອງກັນກັບສັນຍາລັກຫຼັກຊັບນີ້.
2. ສ້າງວັດຖຸ KTable ທີ່ອ່ານຂໍ້ມູນຈາກຫົວຂໍ້ທີ່ມີຂ່າວການແລກປ່ຽນຫຼັກຊັບ. KTable ໃຫມ່ນີ້ຈະຖືກຈັດປະເພດຕາມຂະແຫນງອຸດສາຫະກໍາ.
3. ເຊື່ອມຕໍ່ການອັບເດດຂ່າວທີ່ມີຂໍ້ມູນກ່ຽວກັບຈໍານວນຂອງການແລກປ່ຽນຫຼັກຊັບໂດຍຂະແຫນງອຸດສາຫະກໍາ.

ຕອນນີ້ໃຫ້ເຮົາມາເບິ່ງວິທີປະຕິບັດແຜນປະຕິບັດງານນີ້.

ປ່ຽນ KTable ເປັນ KStream

ເພື່ອປ່ຽນ KTable ເປັນ KStream ທ່ານຕ້ອງເຮັດສິ່ງຕໍ່ໄປນີ້.

1. ໂທຫາວິທີການ KTable.toStream().
2. ໂດຍການໂທຫາວິທີການ KStream.map, ແທນທີ່ກະແຈດ້ວຍຊື່ອຸດສາຫະກໍາ, ແລະຫຼັງຈາກນັ້ນດຶງເອົາວັດຖຸ TransactionSummary ຈາກຕົວຢ່າງ Windowed.

ພວກເຮົາຈະຕ່ອງໂສ້ການດໍາເນີນການເຫຼົ່ານີ້ຮ່ວມກັນດັ່ງຕໍ່ໄປນີ້ (ລະຫັດສາມາດພົບໄດ້ໃນໄຟລ໌ src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (ລາຍການ 5.8).

ເນື່ອງຈາກພວກເຮົາກຳລັງດຳເນີນການ KStream.map, ຕົວຢ່າງ KStream ທີ່ສົ່ງມານັ້ນຖືກແບ່ງແຍກຄືນໃໝ່ໂດຍອັດຕະໂນມັດເມື່ອມັນຖືກໃຊ້ໃນການເຊື່ອມຕໍ່.

ພວກເຮົາໄດ້ສໍາເລັດຂະບວນການປ່ຽນໃຈເຫລື້ອມໃສ, ຕໍ່ໄປພວກເຮົາຈໍາເປັນຕ້ອງສ້າງວັດຖຸ KTable ສໍາລັບການອ່ານຂ່າວຫຼັກຊັບ.

ການສ້າງ KTable ສໍາລັບຂ່າວຫຼັກຊັບ

ໂຊກດີ, ການສ້າງວັດຖຸ KTable ໃຊ້ເວລາພຽງເສັ້ນດຽວຂອງລະຫັດ (ລະຫັດສາມາດພົບໄດ້ໃນ src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (Listing 5.9).

ມັນເປັນມູນຄ່າທີ່ສັງເກດວ່າບໍ່ມີຄວາມຈໍາເປັນທີ່ຈະລະບຸວັດຖຸ Serde, ເພາະວ່າ string Serdes ຖືກນໍາໃຊ້ໃນການຕັ້ງຄ່າ. ນອກຈາກນັ້ນ, ໂດຍໃຊ້ການນັບຈຳ ນວນຄັ້ງທຳອິດ, ຕາຕະລາງແມ່ນເຕັມໄປດ້ວຍບັນທຶກໃນຕອນເລີ່ມຕົ້ນ.

ໃນປັດຈຸບັນພວກເຮົາສາມາດກ້າວໄປສູ່ຂັ້ນຕອນສຸດທ້າຍ - ການເຊື່ອມຕໍ່.

ການເຊື່ອມຕໍ່ການອັບເດດຂ່າວທີ່ມີຂໍ້ມູນຈໍານວນການເຮັດທຸລະກໍາ

ການສ້າງການເຊື່ອມຕໍ່ບໍ່ຍາກ. ພວກເຮົາຈະໃຊ້ການເຂົ້າຮ່ວມຊ້າຍໃນກໍລະນີທີ່ບໍ່ມີຂ່າວຫຼັກຊັບສໍາລັບອຸດສາຫະກໍາທີ່ກ່ຽວຂ້ອງ (ລະຫັດທີ່ຈໍາເປັນສາມາດພົບໄດ້ໃນໄຟລ໌ src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java) (ລາຍຊື່ 5.10).

ຜູ້ປະກອບການ leftJoin ນີ້ແມ່ນຂ້ອນຂ້າງງ່າຍດາຍ. ບໍ່ຄືກັບການເຂົ້າຮ່ວມໃນບົດທີ 4, ວິທີການ JoinWindow ບໍ່ໄດ້ຖືກນໍາໃຊ້ເພາະວ່າເມື່ອປະຕິບັດການເຂົ້າຮ່ວມ KStream-KTable, ມີພຽງແຕ່ຫນຶ່ງເຂົ້າຢູ່ໃນ KTable ສໍາລັບແຕ່ລະກະແຈ. ການເຊື່ອມຕໍ່ດັ່ງກ່າວບໍ່ຈໍາກັດໃນເວລາ: ບັນທຶກແມ່ນຢູ່ໃນ KTable ຫຼືບໍ່ມີ. ການສະຫລຸບຕົ້ນຕໍ: ການນໍາໃຊ້ວັດຖຸ KTable ທ່ານສາມາດເສີມສ້າງ KStream ດ້ວຍຂໍ້ມູນອ້າງອີງທີ່ຖືກປັບປຸງຫນ້ອຍລົງເລື້ອຍໆ.

ຕອນນີ້ພວກເຮົາຈະເບິ່ງວິທີທີ່ມີປະສິດທິພາບຫຼາຍຂຶ້ນເພື່ອເສີມສ້າງກິດຈະກໍາຈາກ KStream.

5.3.4. ວັດຖຸ GlobalKTable

ດັ່ງທີ່ເຈົ້າສາມາດເຫັນໄດ້, ມີຄວາມຕ້ອງການທີ່ຈະເສີມສ້າງກະແສເຫດການຫຼືເພີ່ມບໍລິບົດໃຫ້ພວກເຂົາ. ໃນ​ບົດ​ທີ 4 ທ່ານ​ໄດ້​ເຫັນ​ການ​ເຊື່ອມ​ຕໍ່​ລະ​ຫວ່າງ​ສອງ​ວັດ​ຖຸ KStream, ແລະ​ໃນ​ພາກ​ທີ່​ຜ່ານ​ມາ​ທ່ານ​ໄດ້​ເຫັນ​ການ​ເຊື່ອມ​ຕໍ່​ລະ​ຫວ່າງ KStream ແລະ KTable ໄດ້. ໃນທຸກໆກໍລະນີເຫຼົ່ານີ້, ມັນ ຈຳ ເປັນຕ້ອງແບ່ງສ່ວນກະແສຂໍ້ມູນຄືນ ໃໝ່ ເມື່ອສ້າງແຜນທີ່ຄີກັບປະເພດຫຼືມູນຄ່າ ໃໝ່. ບາງຄັ້ງການແບ່ງສ່ວນແມ່ນເຮັດຢ່າງຈະແຈ້ງ, ແລະບາງຄັ້ງ Kafka Stream ເຮັດມັນໂດຍອັດຕະໂນມັດ. Re-partitioning ແມ່ນມີຄວາມຈໍາເປັນເນື່ອງຈາກວ່າກະແຈໄດ້ມີການປ່ຽນແປງແລະການບັນທຶກຕ້ອງສິ້ນສຸດລົງໃນພາກໃຫມ່, ຖ້າບໍ່ດັ່ງນັ້ນການເຊື່ອມຕໍ່ຈະເປັນໄປບໍ່ໄດ້ (ອັນນີ້ໄດ້ຖືກສົນທະນາໃນບົດທີ 4, ໃນພາກ "ຂໍ້ມູນການແບ່ງສ່ວນໃຫມ່" ໃນພາກຍ່ອຍ 4.2.4).

ການແບ່ງສ່ວນຄືນໃໝ່ມີຄ່າໃຊ້ຈ່າຍ

Re-partitioning ຮຽກຮ້ອງໃຫ້ມີຄ່າໃຊ້ຈ່າຍ - ຄ່າໃຊ້ຈ່າຍຊັບພະຍາກອນເພີ່ມເຕີມສໍາລັບການສ້າງຫົວຂໍ້ລະດັບປານກາງ, ການເກັບຮັກສາຂໍ້ມູນທີ່ຊ້ໍາກັນໃນຫົວຂໍ້ອື່ນ; ມັນຍັງຫມາຍເຖິງການຊັກຊ້າເພີ່ມຂຶ້ນເນື່ອງຈາກການຂຽນແລະການອ່ານຈາກຫົວຂໍ້ນີ້. ນອກຈາກນັ້ນ, ຖ້າທ່ານຕ້ອງການເຂົ້າຮ່ວມຫຼາຍກວ່າຫນຶ່ງດ້ານຫຼືມິຕິ, ທ່ານຕ້ອງຕ່ອງໂສ້ການຮ່ວມ, ແຜນທີ່ບັນທຶກດ້ວຍກະແຈໃຫມ່, ແລະດໍາເນີນການຂະບວນການແບ່ງສ່ວນໃຫມ່ອີກເທື່ອຫນຶ່ງ.

ກຳລັງເຊື່ອມຕໍ່ກັບຊຸດຂໍ້ມູນນ້ອຍກວ່າ

ໃນບາງກໍລະນີ, ປະລິມານຂໍ້ມູນອ້າງອີງທີ່ຈະເຊື່ອມຕໍ່ແມ່ນຂ້ອນຂ້າງນ້ອຍ, ສະນັ້ນການສໍາເນົາທີ່ສົມບູນຂອງມັນສາມາດເຫມາະກັບທ້ອງຖິ່ນໃນແຕ່ລະ node. ສໍາລັບສະຖານະການເຊັ່ນນີ້, Kafka Streams ໃຫ້ຫ້ອງຮຽນ GlobalKTable.

GlobalKTable instances ແມ່ນເປັນເອກະລັກເພາະວ່າແອັບພລິເຄຊັນຈະຈໍາລອງຂໍ້ມູນທັງຫມົດໄປຫາແຕ່ລະ nodes. ແລະເນື່ອງຈາກວ່າຂໍ້ມູນທັງຫມົດແມ່ນມີຢູ່ໃນແຕ່ລະ node, ບໍ່ຈໍາເປັນຕ້ອງແບ່ງສ່ວນກະແສເຫດການໂດຍກະແຈຂໍ້ມູນອ້າງອີງເພື່ອໃຫ້ມັນມີຢູ່ໃນທຸກພາທິຊັນ. ນອກນັ້ນທ່ານຍັງສາມາດເຮັດການເຂົ້າຮ່ວມ keyless ໂດຍໃຊ້ວັດຖຸ GlobalKTable. ໃຫ້ກັບຄືນໄປຫາຫນຶ່ງໃນຕົວຢ່າງທີ່ຜ່ານມາເພື່ອສະແດງໃຫ້ເຫັນຄຸນສົມບັດນີ້.

ການເຊື່ອມຕໍ່ວັດຖຸ KStream ກັບວັດຖຸ GlobalKTable

ໃນພາກຍ່ອຍ 5.3.2, ພວກເຮົາໄດ້ປະຕິບັດການລວບລວມປ່ອງຢ້ຽມຂອງທຸລະກໍາແລກປ່ຽນໂດຍຜູ້ຊື້. ຜົນໄດ້ຮັບຂອງການລວບລວມນີ້ເບິ່ງຄືດັ່ງນີ້:

{customerId='074-09-3705', stockTicker='GUTM'}, 17
{customerId='037-34-5184', stockTicker='CORK'}, 16

ໃນຂະນະທີ່ຜົນໄດ້ຮັບເຫຼົ່ານີ້ໄດ້ຮັບໃຊ້ຈຸດປະສົງ, ມັນຈະເປັນປະໂຫຍດກວ່າຖ້າຊື່ຂອງລູກຄ້າແລະຊື່ເຕັມຂອງບໍລິສັດໄດ້ຖືກສະແດງ. ເພື່ອເພີ່ມຊື່ລູກຄ້າແລະຊື່ບໍລິສັດ, ທ່ານສາມາດເຮັດການເຂົ້າຮ່ວມແບບປົກກະຕິ, ແຕ່ທ່ານຈະຕ້ອງເຮັດສອງແຜນທີ່ທີ່ສໍາຄັນແລະການແບ່ງສ່ວນໃຫມ່. ດ້ວຍ GlobalKTable ທ່ານສາມາດຫລີກລ້ຽງຄ່າໃຊ້ຈ່າຍຂອງການດໍາເນີນງານດັ່ງກ່າວ.

ເພື່ອເຮັດສິ່ງນີ້, ພວກເຮົາຈະໃຊ້ countStream object ຈາກ Listing 5.11 (ລະຫັດທີ່ສອດຄ້ອງກັນສາມາດພົບໄດ້ໃນ src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) ແລະເຊື່ອມຕໍ່ມັນກັບສອງວັດຖຸ GlobalKTable.

ພວກເຮົາໄດ້ສົນທະນາເລື່ອງນີ້ກ່ອນແລ້ວ, ສະນັ້ນຂ້ອຍຈະບໍ່ເວົ້າຊ້ຳອີກ ແຕ່ຂ້ອຍສັງເກດວ່າລະຫັດໃນຟັງຊັນ toStream().map ແມ່ນ abstracted ເຂົ້າໄປໃນ function object ແທນທີ່ຈະເປັນ inline lambda expression ເພື່ອປະໂຫຍດຂອງການອ່ານໄດ້.

ຂັ້ນຕອນຕໍ່ໄປແມ່ນການປະກາດສອງຕົວຢ່າງຂອງ GlobalKTable (ລະຫັດທີ່ສະແດງສາມາດພົບໄດ້ໃນໄຟລ໌ src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (ລາຍການ 5.12).

ກະ​ລຸ​ນາ​ສັງ​ເກດ​ວ່າ​ຊື່​ຫົວ​ຂໍ້​ແມ່ນ​ໄດ້​ອະ​ທິ​ບາຍ​ໂດຍ​ການ​ນໍາ​ໃຊ້​ປະ​ເພດ enumerated​.

ໃນປັດຈຸບັນທີ່ພວກເຮົາມີອົງປະກອບທັງຫມົດກຽມພ້ອມ, ທັງຫມົດທີ່ຍັງເຫຼືອແມ່ນການຂຽນລະຫັດສໍາລັບການເຊື່ອມຕໍ່ (ເຊິ່ງສາມາດພົບໄດ້ໃນໄຟລ໌ src/main/java/bbejeck/chapter_5/GlobalKTableExample.java) (ລາຍຊື່ 5.13).

ເຖິງແມ່ນວ່າມີສອງຕົວເຂົ້າຮ່ວມໃນລະຫັດນີ້, ພວກມັນຖືກຕ່ອງໂສ້ເພາະວ່າທັງສອງຜົນໄດ້ຮັບຂອງພວກມັນບໍ່ໄດ້ຖືກນໍາໃຊ້ແຍກຕ່າງຫາກ. ຜົນໄດ້ຮັບແມ່ນສະແດງໃນຕອນທ້າຍຂອງການດໍາເນີນງານທັງຫມົດ.

ເມື່ອ​ທ່ານ​ດໍາ​ເນີນ​ການ​ປະ​ຕິ​ບັດ​ການ​ເຂົ້າ​ຮ່ວມ​ຂ້າງ​ເທິງ​ນີ້​, ທ່ານ​ຈະ​ໄດ້​ຮັບ​ຜົນ​ໄດ້​ຮັບ​ດັ່ງ​ນີ້​:

{customer='Barney, Smith' company="Exxon", transactions= 17}

ໂດຍເນື້ອແທ້ແລ້ວບໍ່ໄດ້ປ່ຽນແປງ, ແຕ່ຜົນໄດ້ຮັບເຫຼົ່ານີ້ເບິ່ງທີ່ຊັດເຈນກວ່າ.

ຖ້າທ່ານນັບລົງໄປໃນບົດທີ 4, ທ່ານໄດ້ພົບເຫັນການເຊື່ອມຕໍ່ຫຼາຍປະເພດໃນການປະຕິບັດ. ພວກເຂົາຖືກລະບຸໄວ້ໃນຕາຕະລາງ. 5.2. ຕາຕະລາງນີ້ສະທ້ອນເຖິງຄວາມສາມາດໃນການເຊື່ອມຕໍ່ເປັນສະບັບ 1.0.0 ຂອງ Kafka Stream; ບາງສິ່ງບາງຢ່າງອາດຈະມີການປ່ຽນແປງໃນການປ່ອຍໃນອະນາຄົດ.

ເພື່ອສະຫຼຸບສິ່ງຕ່າງໆ, ໃຫ້ພວກເຮົາສະຫຼຸບພື້ນຖານ: ທ່ານສາມາດເຊື່ອມຕໍ່ການຖ່າຍທອດເຫດການ (KStream) ແລະອັບເດດການຖ່າຍທອດ (KTable) ໂດຍໃຊ້ສະຖານະທ້ອງຖິ່ນ. ອີກທາງເລືອກ, ຖ້າຂະຫນາດຂອງຂໍ້ມູນອ້າງອີງບໍ່ໃຫຍ່ເກີນໄປ, ທ່ານສາມາດນໍາໃຊ້ວັດຖຸ GlobalKTable. GlobalKTables ເຮັດເລື້ມຄືນພາທິຊັນທັງໝົດໃສ່ແຕ່ລະໂຫນດແອັບພລິເຄຊັນ Kafka Stream, ຮັບປະກັນວ່າຂໍ້ມູນທັງໝົດມີຢູ່ ໂດຍບໍ່ຄໍານຶງວ່າການແບ່ງພາທິຊັນຈະກົງກັບລະຫັດໃດ.

ຕໍ່ໄປພວກເຮົາຈະເຫັນຄຸນສົມບັດ Kafka Stream, ຂອບໃຈທີ່ພວກເຮົາສາມາດສັງເກດເຫັນການປ່ຽນແປງຂອງລັດໂດຍບໍ່ມີການບໍລິໂພກຂໍ້ມູນຈາກຫົວຂໍ້ Kafka.

5.3.5. ສະຖານະທີ່ສາມາດສອບຖາມໄດ້

ພວກເຮົາໄດ້ປະຕິບັດການດໍາເນີນການຈໍານວນຫນຶ່ງທີ່ກ່ຽວຂ້ອງກັບລັດແລະສະເຫມີໃຫ້ຜົນໄດ້ຮັບກັບ console (ສໍາລັບຈຸດປະສົງການພັດທະນາ) ຫຼືຂຽນໃຫ້ເຂົາເຈົ້າໃນຫົວຂໍ້ໃດຫນຶ່ງ (ສໍາລັບຈຸດປະສົງການຜະລິດ). ເມື່ອຂຽນຜົນໄດ້ຮັບໄປຫາຫົວຂໍ້ໃດຫນຶ່ງ, ທ່ານຕ້ອງໃຊ້ຜູ້ບໍລິໂພກ Kafka ເພື່ອເບິ່ງພວກມັນ.

ການອ່ານຂໍ້ມູນຈາກຫົວຂໍ້ເຫຼົ່ານີ້ສາມາດຖືວ່າເປັນປະເພດຂອງການເບິ່ງທີ່ເປັນວັດຖຸ. ສໍາລັບຈຸດປະສົງຂອງພວກເຮົາ, ພວກເຮົາສາມາດໃຊ້ຄໍານິຍາມຂອງມຸມເບິ່ງທີ່ເປັນວັດຖຸຈາກວິກິພີເດຍ: "... ວັດຖຸຖານຂໍ້ມູນທີ່ມີຜົນມາຈາກການສອບຖາມ. ສໍາລັບຕົວຢ່າງ, ມັນສາມາດເປັນສໍາເນົາທ້ອງຖິ່ນຂອງຂໍ້ມູນຫ່າງໄກສອກຫຼີກ, ຫຼືຊຸດຍ່ອຍຂອງແຖວແລະ / ຫຼືຄໍລໍາຂອງຕາຕະລາງຫຼືຜົນການເຂົ້າຮ່ວມ, ຫຼືຕາຕະລາງສະຫຼຸບທີ່ໄດ້ຮັບໂດຍຜ່ານການລວບລວມ" (https://en.wikipedia.org/wiki /materialized_view).

Kafka Streams ຍັງຊ່ວຍໃຫ້ທ່ານສາມາດດໍາເນີນການສອບຖາມແບບໂຕ້ຕອບໃນຮ້ານຄ້າຂອງລັດ, ຊ່ວຍໃຫ້ທ່ານສາມາດອ່ານຄວາມຄິດເຫັນທີ່ເປັນວັດຖຸເຫຼົ່ານີ້ໂດຍກົງ. ມັນເປັນສິ່ງສໍາຄັນທີ່ຈະສັງເກດວ່າການສອບຖາມກັບຮ້ານຂອງລັດແມ່ນການດໍາເນີນງານແບບອ່ານເທົ່ານັ້ນ. ນີ້ຮັບປະກັນວ່າທ່ານບໍ່ຈໍາເປັນຕ້ອງກັງວົນກ່ຽວກັບການເຮັດໃຫ້ລັດບໍ່ສອດຄ່ອງໂດຍບັງເອີນໃນຂະນະທີ່ຄໍາຮ້ອງສະຫມັກຂອງທ່ານກໍາລັງປະມວນຜົນຂໍ້ມູນ.

ຄວາມສາມາດໃນການສອບຖາມໂດຍກົງກັບຮ້ານຄ້າຂອງລັດແມ່ນສໍາຄັນ. ນີ້ຫມາຍຄວາມວ່າທ່ານສາມາດສ້າງຄໍາຮ້ອງສະຫມັກ dashboard ໂດຍບໍ່ມີການທໍາອິດທີ່ຈະດຶງຂໍ້ມູນຈາກຜູ້ບໍລິໂພກ Kafka. ມັນຍັງເພີ່ມປະສິດທິພາບຂອງຄໍາຮ້ອງສະຫມັກ, ເນື່ອງຈາກຄວາມຈິງທີ່ວ່າບໍ່ຈໍາເປັນຕ້ອງຂຽນຂໍ້ມູນອີກເທື່ອຫນຶ່ງ:

• ຂໍຂອບໃຈກັບທ້ອງຖິ່ນຂອງຂໍ້ມູນ, ເຂົາເຈົ້າສາມາດເຂົ້າເຖິງໄດ້ໄວ;
• ການຊໍ້າຊ້ອນຂອງຂໍ້ມູນຖືກລົບລ້າງ, ເພາະວ່າມັນບໍ່ໄດ້ຂຽນໃສ່ບ່ອນເກັບຂໍ້ມູນພາຍນອກ.

ສິ່ງຕົ້ນຕໍທີ່ຂ້ອຍຕ້ອງການໃຫ້ເຈົ້າຈື່ແມ່ນວ່າເຈົ້າສາມາດສອບຖາມໂດຍກົງຈາກພາຍໃນແອັບພລິເຄຊັນຂອງເຈົ້າ. ໂອກາດນີ້ເຮັດໃຫ້ທ່ານບໍ່ສາມາດເວົ້າເກີນໄດ້. ແທນທີ່ຈະບໍລິໂພກຂໍ້ມູນຈາກ Kafka ແລະເກັບຮັກສາບັນທຶກໃນຖານຂໍ້ມູນສໍາລັບແອັບພລິເຄຊັນ, ທ່ານສາມາດສອບຖາມຮ້ານຄ້າຂອງລັດດ້ວຍຜົນໄດ້ຮັບດຽວກັນ. ການສອບຖາມໂດຍກົງກັບຮ້ານຄ້າຂອງລັດຫມາຍຄວາມວ່າລະຫັດຫນ້ອຍ (ບໍ່ມີຜູ້ບໍລິໂພກ) ແລະຊອບແວຫນ້ອຍ (ບໍ່ຈໍາເປັນຕ້ອງມີຕາຕະລາງຖານຂໍ້ມູນເພື່ອເກັບຜົນໄດ້ຮັບ).

ພວກເຮົາໄດ້ກວມເອົາພື້ນຖານເລັກນ້ອຍໃນບົດນີ້, ດັ່ງນັ້ນພວກເຮົາຈະອອກຈາກການສົນທະນາຂອງພວກເຮົາກ່ຽວກັບການສອບຖາມແບບໂຕ້ຕອບຕໍ່ກັບຮ້ານຄ້າຂອງລັດໃນປັດຈຸບັນ. ແຕ່ຢ່າກັງວົນ: ໃນບົດທີ 9, ພວກເຮົາຈະສ້າງຄໍາຮ້ອງສະຫມັກ dashboard ງ່າຍດາຍທີ່ມີການສອບຖາມແບບໂຕ້ຕອບ. ມັນຈະນໍາໃຊ້ບາງຕົວຢ່າງຈາກບົດນີ້ແລະບົດທີ່ຜ່ານມາເພື່ອສະແດງໃຫ້ເຫັນການສອບຖາມແບບໂຕ້ຕອບແລະວິທີທີ່ທ່ານສາມາດເພີ່ມພວກມັນໃສ່ຄໍາຮ້ອງສະຫມັກ Kafka Stream.

Summary

• ວັດຖຸ KStream ເປັນຕົວແທນສະຕຣີມຂອງເຫດການ, ປຽບທຽບກັບການແຊກໃສ່ຖານຂໍ້ມູນ. ວັດຖຸ KTable ສະແດງເຖິງການຖ່າຍທອດການອັບເດດ, ຄືກັບການອັບເດດຖານຂໍ້ມູນ. ຂະຫນາດຂອງວັດຖຸ KTable ບໍ່ເຕີບໂຕ, ບັນທຶກເກົ່າຖືກແທນທີ່ດ້ວຍອັນໃຫມ່.
• ຕ້ອງການວັດຖຸ KTable ສໍາລັບການດໍາເນີນງານການລວບລວມ.
• ການນໍາໃຊ້ການດໍາເນີນງານ windowing, ທ່ານສາມາດແບ່ງປັນຂໍ້ມູນລວມເຂົ້າໄປໃນຖັງເວລາ.
• ຂໍຂອບໃຈກັບວັດຖຸ GlobalKTable, ທ່ານສາມາດເຂົ້າເຖິງຂໍ້ມູນອ້າງອີງທຸກບ່ອນໃນແອັບພລິເຄຊັນ, ບໍ່ວ່າຈະເປັນການແບ່ງສ່ວນ.
• ການເຊື່ອມຕໍ່ລະຫວ່າງວັດຖຸ KStream, KTable ແລະ GlobalKTable ແມ່ນເປັນໄປໄດ້.

ມາຮອດປະຈຸ, ພວກເຮົາໄດ້ສຸມໃສ່ການສ້າງຄໍາຮ້ອງສະຫມັກ Kafka Stream ໂດຍໃຊ້ KStream DSL ລະດັບສູງ. ເຖິງແມ່ນວ່າວິທີການລະດັບສູງຊ່ວຍໃຫ້ທ່ານສ້າງໂຄງການທີ່ຊັດເຈນແລະຊັດເຈນ, ການນໍາໃຊ້ມັນເປັນຕົວແທນຂອງການຄ້າ. ການເຮັດວຽກກັບ DSL KStream ຫມາຍເຖິງການເພີ່ມຄວາມສອດຄ່ອງຂອງລະຫັດຂອງທ່ານໂດຍການຫຼຸດຜ່ອນລະດັບການຄວບຄຸມ. ໃນບົດຕໍ່ໄປ, ພວກເຮົາຈະເບິ່ງ API node handler ລະດັບຕ່ໍາແລະລອງການຄ້າອື່ນໆ. ໂປຣແກມຕ່າງໆຈະຍາວກວ່າທີ່ເຄີຍເປັນມາ, ແຕ່ພວກເຮົາຈະສາມາດສ້າງ node handler ເກືອບທຸກອັນທີ່ເຮົາອາດຈະຕ້ອງການ.

→ ລາຍລະອຽດເພີ່ມເຕີມກ່ຽວກັບປຶ້ມສາມາດເຂົ້າໄປເບິ່ງໄດ້ທີ່ ເວັບໄຊທ໌ຂອງຜູ້ເຜີຍແຜ່

→ ສໍາລັບ Habrozhiteli ສ່ວນຫຼຸດ 25% ໂດຍໃຊ້ຄູປ໋ອງ - ກະແສ Kafka

→ ເມື່ອຈ່າຍເງິນໃຫ້ກັບສະບັບເຈ້ຍຂອງປຶ້ມ, ປຶ້ມອີເລັກໂທຣນິກຈະຖືກສົ່ງໄປທາງອີເມລ.

ແຫຼ່ງຂໍ້ມູນ: www.habr.com