"കാഫ്ക സ്ട്രീംസ് ഇൻ ആക്ഷൻ" എന്ന പുസ്തകം. തത്സമയ ജോലിക്കുള്ള ആപ്ലിക്കേഷനുകളും മൈക്രോ സർവീസുകളും"

ഹലോ, ഖബ്രോ നിവാസികൾ! ത്രെഡ് പ്രോസസ്സിംഗ് മനസ്സിലാക്കാൻ ആഗ്രഹിക്കുന്ന ഏതൊരു ഡവലപ്പർക്കും ഈ പുസ്തകം അനുയോജ്യമാണ്. വിതരണം ചെയ്ത പ്രോഗ്രാമിംഗ് മനസ്സിലാക്കുന്നത് കാഫ്ക, കാഫ്ക സ്ട്രീമുകൾ നന്നായി മനസ്സിലാക്കാൻ നിങ്ങളെ സഹായിക്കും. കാഫ്ക ചട്ടക്കൂട് അറിയുന്നത് നന്നായിരിക്കും, പക്ഷേ ഇത് ആവശ്യമില്ല: നിങ്ങൾക്ക് ആവശ്യമുള്ളതെല്ലാം ഞാൻ നിങ്ങളോട് പറയും. പരിചയസമ്പന്നരായ കാഫ്ക ഡെവലപ്പർമാരും തുടക്കക്കാരും ഈ പുസ്തകത്തിൽ കാഫ്ക സ്ട്രീംസ് ലൈബ്രറി ഉപയോഗിച്ച് രസകരമായ സ്ട്രീം പ്രോസസ്സിംഗ് ആപ്ലിക്കേഷനുകൾ എങ്ങനെ സൃഷ്ടിക്കാമെന്ന് പഠിക്കും. സീരിയലൈസേഷൻ പോലുള്ള ആശയങ്ങളുമായി ഇതിനകം പരിചിതരായ ഇന്റർമീഡിയറ്റ്, അഡ്വാൻസ്ഡ് ജാവ ഡെവലപ്പർമാർ തങ്ങളുടെ കഴിവുകൾ കാഫ്ക സ്ട്രീം ആപ്ലിക്കേഷനുകൾ സൃഷ്ടിക്കാൻ പഠിക്കും. പുസ്തകത്തിന്റെ സോഴ്സ് കോഡ് Java 8-ൽ എഴുതിയിരിക്കുന്നു, കൂടാതെ Java 8 lambda എക്സ്പ്രഷൻ സിന്റാക്സ് ഗണ്യമായി ഉപയോഗിക്കുന്നു, അതിനാൽ ലാംഡ ഫംഗ്ഷനുകൾ (മറ്റൊരു പ്രോഗ്രാമിംഗ് ഭാഷയിൽ പോലും) എങ്ങനെ പ്രവർത്തിക്കണമെന്ന് അറിയുന്നത് ഉപയോഗപ്രദമാകും.

ഉദ്ധരണി. 5.3 അഗ്രഗേഷൻ, വിൻഡോ ചെയ്യൽ പ്രവർത്തനങ്ങൾ

ഈ വിഭാഗത്തിൽ, കാഫ്ക സ്ട്രീമുകളുടെ ഏറ്റവും പ്രതീക്ഷ നൽകുന്ന ഭാഗങ്ങൾ പര്യവേക്ഷണം ചെയ്യാൻ ഞങ്ങൾ മുന്നോട്ട് പോകും. കാഫ്ക സ്ട്രീമുകളുടെ ഇനിപ്പറയുന്ന വശങ്ങൾ ഞങ്ങൾ ഇതുവരെ ഉൾപ്പെടുത്തിയിട്ടുണ്ട്:

• ഒരു പ്രോസസ്സിംഗ് ടോപ്പോളജി സൃഷ്ടിക്കുന്നു;
• സ്ട്രീമിംഗ് ആപ്ലിക്കേഷനുകളിൽ സ്റ്റേറ്റ് ഉപയോഗിക്കുന്നത്;
• ഡാറ്റ സ്ട്രീം കണക്ഷനുകൾ നടത്തുന്നു;
• ഇവന്റ് സ്ട്രീമുകളും (KStream) അപ്ഡേറ്റ് സ്ട്രീമുകളും (KTable) തമ്മിലുള്ള വ്യത്യാസങ്ങൾ.

ഇനിപ്പറയുന്ന ഉദാഹരണങ്ങളിൽ ഞങ്ങൾ ഈ ഘടകങ്ങളെല്ലാം ഒരുമിച്ച് കൊണ്ടുവരും. സ്ട്രീമിംഗ് ആപ്ലിക്കേഷനുകളുടെ മറ്റൊരു മികച്ച സവിശേഷതയായ വിൻഡോയിംഗിനെക്കുറിച്ച് നിങ്ങൾ പഠിക്കും. ഞങ്ങളുടെ ആദ്യ ഉദാഹരണം ഒരു ലളിതമായ സമാഹാരമായിരിക്കും.

5.3.1. വ്യവസായ മേഖലയുടെ അടിസ്ഥാനത്തിൽ ഓഹരി വിൽപ്പനയുടെ സമാഹാരം

സ്ട്രീമിംഗ് ഡാറ്റയുമായി പ്രവർത്തിക്കുമ്പോൾ അഗ്രഗേഷനും ഗ്രൂപ്പിംഗും സുപ്രധാന ഉപകരണങ്ങളാണ്. വ്യക്തിഗത രേഖകൾ ലഭിക്കുന്നതിനനുസരിച്ച് പരിശോധിക്കുന്നത് പലപ്പോഴും അപര്യാപ്തമാണ്. ഡാറ്റയിൽ നിന്ന് കൂടുതൽ വിവരങ്ങൾ എക്‌സ്‌ട്രാക്റ്റുചെയ്യുന്നതിന്, അവയെ ഗ്രൂപ്പുചെയ്യുകയും സംയോജിപ്പിക്കുകയും ചെയ്യേണ്ടത് ആവശ്യമാണ്.

ഈ ഉദാഹരണത്തിൽ, നിരവധി വ്യവസായങ്ങളിലെ കമ്പനികളുടെ സ്റ്റോക്കുകളുടെ വിൽപ്പന അളവ് ട്രാക്ക് ചെയ്യേണ്ട ഒരു ഡേ ട്രേഡറുടെ വേഷം നിങ്ങൾ ധരിക്കും. പ്രത്യേകമായി, ഓരോ വ്യവസായത്തിലും ഏറ്റവും വലിയ ഓഹരി വിൽപ്പനയുള്ള അഞ്ച് കമ്പനികളിൽ നിങ്ങൾക്ക് താൽപ്പര്യമുണ്ട്.

അത്തരം സമാഹരണത്തിന് ആവശ്യമായ ഫോമിലേക്ക് ഡാറ്റ വിവർത്തനം ചെയ്യുന്നതിന് ഇനിപ്പറയുന്ന നിരവധി ഘട്ടങ്ങൾ ആവശ്യമായി വരും (സാമാന്യമായി പറഞ്ഞാൽ).

1. അസംസ്‌കൃത സ്റ്റോക്ക് ട്രേഡിംഗ് വിവരങ്ങൾ പ്രസിദ്ധീകരിക്കുന്ന വിഷയാധിഷ്ഠിത ഉറവിടം സൃഷ്‌ടിക്കുക. സ്റ്റോക്ക് ട്രാൻസാക്ഷൻ എന്ന തരത്തിലുള്ള ഒരു ഒബ്‌ജക്റ്റ് ഷെയർ വോളിയം തരത്തിന്റെ ഒബ്‌ജക്റ്റിലേക്ക് ഞങ്ങൾ മാപ്പ് ചെയ്യേണ്ടതുണ്ട്. StockTransaction ഒബ്‌ജക്റ്റിൽ സെയിൽസ് മെറ്റാഡാറ്റ അടങ്ങിയിരിക്കുന്നു എന്നതാണ് കാര്യം, എന്നാൽ ഞങ്ങൾക്ക് വിൽക്കുന്ന ഷെയറുകളുടെ എണ്ണത്തെക്കുറിച്ചുള്ള ഡാറ്റ മാത്രമേ ആവശ്യമുള്ളൂ.
2. സ്റ്റോക്ക് ചിഹ്നമനുസരിച്ച് ഷെയർ വോളിയം ഡാറ്റ ഗ്രൂപ്പ് ചെയ്യുക. ചിഹ്നമനുസരിച്ച് ഗ്രൂപ്പുചെയ്‌തുകഴിഞ്ഞാൽ, നിങ്ങൾക്ക് ഈ ഡാറ്റ സ്റ്റോക്ക് സെയിൽസ് വോള്യങ്ങളുടെ സബ്‌ടോട്ടലുകളായി ചുരുക്കാനാകും. KStream.groupBy രീതി KGroupedStream തരത്തിന്റെ ഒരു ഉദാഹരണം നൽകുന്നു എന്നത് ശ്രദ്ധിക്കേണ്ടതാണ്. KGroupedStream.reduce മെത്തേഡിലേക്ക് കൂടുതൽ വിളിക്കുന്നതിലൂടെ നിങ്ങൾക്ക് ഒരു KTable ഉദാഹരണം ലഭിക്കും.

എന്താണ് KGroupedStream ഇന്റർഫേസ്

KStream.groupBy, KStream.groupByKey രീതികൾ KGroupedStream-ന്റെ ഒരു ഉദാഹരണം നൽകുന്നു. KGroupedStream എന്നത് കീകൾ പ്രകാരം ഗ്രൂപ്പുചെയ്‌തതിന് ശേഷം ഇവന്റുകളുടെ ഒരു സ്ട്രീമിന്റെ ഇന്റർമീഡിയറ്റ് പ്രാതിനിധ്യമാണ്. ഇത് നേരിട്ട് പ്രവർത്തിക്കാൻ ഉദ്ദേശിച്ചുള്ളതല്ല. പകരം, അഗ്രഗേഷൻ പ്രവർത്തനങ്ങൾക്കായി KGroupedStream ഉപയോഗിക്കുന്നു, ഇത് എല്ലായ്പ്പോഴും ഒരു KTable-ൽ കലാശിക്കുന്നു. അഗ്രഗേഷൻ പ്രവർത്തനങ്ങളുടെ ഫലം ഒരു KTable ആയതിനാലും അവർ ഒരു സ്റ്റേറ്റ് സ്റ്റോർ ഉപയോഗിക്കുന്നതിനാലും, ഫലമായുണ്ടാകുന്ന എല്ലാ അപ്‌ഡേറ്റുകളും പൈപ്പ് ലൈനിലൂടെ കൂടുതൽ താഴേക്ക് അയയ്‌ക്കാതിരിക്കാൻ സാധ്യതയുണ്ട്.

KTable.groupBy രീതി സമാനമായ KGroupedTable നൽകുന്നു - കീ ഉപയോഗിച്ച് വീണ്ടും ഗ്രൂപ്പുചെയ്‌ത അപ്‌ഡേറ്റുകളുടെ സ്ട്രീമിന്റെ ഒരു ഇന്റർമീഡിയറ്റ് പ്രാതിനിധ്യം.

നമുക്ക് ഒരു ചെറിയ ഇടവേള എടുത്ത് ചിത്രം നോക്കാം. 5.9, ഞങ്ങൾ എന്താണ് നേടിയതെന്ന് കാണിക്കുന്നു. ഈ ടോപ്പോളജി നിങ്ങൾക്ക് ഇതിനകം വളരെ പരിചിതമായിരിക്കണം.

നമുക്ക് ഇപ്പോൾ ഈ ടോപ്പോളജിയുടെ കോഡ് നോക്കാം (ഇത് src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java എന്ന ഫയലിൽ കാണാം) (ലിസ്റ്റിംഗ് 5.2).

നൽകിയിരിക്കുന്ന കോഡ് അതിന്റെ സംക്ഷിപ്തതയും നിരവധി വരികളിൽ നടത്തിയ പ്രവർത്തനങ്ങളുടെ വലിയ അളവും കൊണ്ട് വേർതിരിച്ചിരിക്കുന്നു. ബിൽഡർ.സ്ട്രീം രീതിയുടെ ആദ്യ പാരാമീറ്ററിൽ നിങ്ങൾ പുതിയ എന്തെങ്കിലും ശ്രദ്ധിച്ചേക്കാം: എനം തരം AutoOffsetReset.EARLIEST (ഏറ്റവും പുതിയതും ഉണ്ട്), Consumed.withOffsetResetPolicy രീതി ഉപയോഗിച്ച് സജ്ജമാക്കിയ ഒരു മൂല്യം. ഓരോ KStream അല്ലെങ്കിൽ KTable-നും ഒരു ഓഫ്‌സെറ്റ് റീസെറ്റ് സ്ട്രാറ്റജി വ്യക്തമാക്കാൻ ഈ എണ്ണൽ തരം ഉപയോഗിക്കാവുന്നതാണ്, കോൺഫിഗറേഷനിൽ നിന്നുള്ള ഓഫ്‌സെറ്റ് റീസെറ്റ് ഓപ്‌ഷനേക്കാൾ മുൻതൂക്കം എടുക്കുന്നു.

GroupByKey, GroupBy

KStream ഇന്റർഫേസിന് റെക്കോർഡുകൾ ഗ്രൂപ്പുചെയ്യുന്നതിന് രണ്ട് രീതികളുണ്ട്: GroupByKey, GroupBy. രണ്ടും ഒരു KGroupedTable നൽകുന്നു, അതിനാൽ അവ തമ്മിലുള്ള വ്യത്യാസം എന്താണെന്നും ഏത് എപ്പോൾ ഉപയോഗിക്കണമെന്നും നിങ്ങൾ ചിന്തിച്ചേക്കാം?

KStream-ലെ കീകൾ ഇതിനകം ശൂന്യമായിരിക്കുമ്പോൾ GroupByKey രീതി ഉപയോഗിക്കുന്നു. ഏറ്റവും പ്രധാനമായി, "വീണ്ടും വിഭജനം ആവശ്യമാണ്" എന്ന ഫ്ലാഗ് ഒരിക്കലും സജ്ജീകരിച്ചിട്ടില്ല.

നിങ്ങൾ ഗ്രൂപ്പിംഗ് കീകൾ മാറ്റിയതായി GroupBy രീതി അനുമാനിക്കുന്നു, അതിനാൽ റീപാർട്ടീഷൻ ഫ്ലാഗ് true ആയി സജ്ജീകരിച്ചിരിക്കുന്നു. GroupBy രീതിക്ക് ശേഷം ചേരലുകൾ, അഗ്രിഗേഷനുകൾ മുതലായവ നടത്തുന്നത് യാന്ത്രികമായി വീണ്ടും പാർട്ടീഷനിംഗിന് കാരണമാകും.
സംഗ്രഹം: സാധ്യമാകുമ്പോഴെല്ലാം, GroupBy ഉപയോഗിക്കുന്നതിന് പകരം GroupByKey ഉപയോഗിക്കേണ്ടതാണ്.

മാപ്പ് മൂല്യങ്ങളും ഗ്രൂപ്പ് ബൈ രീതികളും എന്താണ് ചെയ്യുന്നതെന്ന് വ്യക്തമാണ്, അതിനാൽ നമുക്ക് സം() രീതി (src/main/java/bbejeck/model/ShareVolume.java-ൽ കാണപ്പെടുന്നു) (ലിസ്റ്റിംഗ് 5.3) നോക്കാം.

ShareVolume.sum രീതി സ്റ്റോക്ക് സെയിൽസ് വോളിയത്തിന്റെ റണ്ണിംഗ് ടോട്ടൽ നൽകുന്നു, കൂടാതെ കണക്കുകൂട്ടലുകളുടെ മുഴുവൻ ശൃംഖലയുടെയും ഫലം ഒരു KTable ഒബ്ജക്റ്റാണ്. . KTable വഹിക്കുന്ന പങ്ക് ഇപ്പോൾ നിങ്ങൾ മനസ്സിലാക്കുന്നു. ShareVolume ഒബ്‌ജക്‌റ്റുകൾ വരുമ്പോൾ, അനുബന്ധ KTable ഒബ്‌ജക്റ്റ് ഏറ്റവും പുതിയ നിലവിലെ അപ്‌ഡേറ്റ് സംഭരിക്കുന്നു. എല്ലാ അപ്‌ഡേറ്റുകളും മുമ്പത്തെ shareVolumeKTable-ൽ പ്രതിഫലിച്ചിട്ടുണ്ടെന്ന കാര്യം ഓർത്തിരിക്കേണ്ടത് പ്രധാനമാണ്, എന്നാൽ എല്ലാം പിന്നീട് അയച്ചിട്ടില്ല.

ഓരോ വ്യവസായത്തിലും ഏറ്റവും കൂടുതൽ ഓഹരികൾ വ്യാപാരം ചെയ്യുന്ന അഞ്ച് കമ്പനികളിൽ എത്തിച്ചേരുന്നതിന് (ട്രേഡ് ചെയ്ത ഷെയറുകളുടെ എണ്ണം അനുസരിച്ച്) ഞങ്ങൾ ഈ KTable ഉപയോഗിക്കുന്നു. ഈ കേസിലെ ഞങ്ങളുടെ പ്രവർത്തനങ്ങൾ ആദ്യ സമാഹരണത്തിന് സമാനമായിരിക്കും.

1. വ്യവസായം അനുസരിച്ച് വ്യക്തിഗത ഷെയർ വോളിയം ഒബ്ജക്റ്റുകൾ ഗ്രൂപ്പുചെയ്യാൻ മറ്റൊരു ഗ്രൂപ്പ് ബൈ ഓപ്പറേഷൻ നടത്തുക.
2. ShareVolume ഒബ്‌ജക്‌റ്റുകൾ സംഗ്രഹിക്കാൻ ആരംഭിക്കുക. ഇത്തവണ അഗ്രഗേഷൻ ഒബ്‌ജക്‌റ്റ് ഒരു നിശ്ചിത വലുപ്പത്തിലുള്ള മുൻഗണനാ ക്യൂവാണ്. ഈ നിശ്ചിത വലുപ്പത്തിലുള്ള ക്യൂവിൽ, ഏറ്റവും കൂടുതൽ ഓഹരികൾ വിറ്റഴിച്ച അഞ്ച് കമ്പനികൾ മാത്രമേ നിലനിർത്തൂ.
3. മുമ്പത്തെ ഖണ്ഡികയിൽ നിന്നുള്ള ക്യൂകൾ ഒരു സ്‌ട്രിംഗ് മൂല്യത്തിലേക്ക് മാപ്പ് ചെയ്യുക, വ്യവസായം അനുസരിച്ച് ഏറ്റവും കൂടുതൽ ട്രേഡ് ചെയ്യപ്പെടുന്ന മികച്ച അഞ്ച് സ്റ്റോക്കുകൾ തിരികെ നൽകുക.
4. വിഷയത്തിൽ ഫലങ്ങൾ സ്ട്രിംഗ് രൂപത്തിൽ എഴുതുക.

ചിത്രത്തിൽ. ചിത്രം 5.10 ഡാറ്റാ ഫ്ലോ ടോപ്പോളജി ഗ്രാഫ് കാണിക്കുന്നു. നിങ്ങൾക്ക് കാണാനാകുന്നതുപോലെ, രണ്ടാം ഘട്ട പ്രോസസ്സിംഗ് വളരെ ലളിതമാണ്.

ഈ രണ്ടാം ഘട്ട പ്രോസസ്സിംഗിന്റെ ഘടനയെക്കുറിച്ച് ഇപ്പോൾ ഞങ്ങൾക്ക് വ്യക്തമായ ധാരണയുണ്ട്, നമുക്ക് അതിന്റെ സോഴ്സ് കോഡിലേക്ക് തിരിയാം (നിങ്ങൾ അത് src/main/java/bbejeck/chapter_5/AggregationsAndReducingExample.java എന്ന ഫയലിൽ കണ്ടെത്തും) (ലിസ്റ്റിംഗ് 5.4) .

ഈ ഇനീഷ്യലൈസറിൽ ഒരു ഫിക്സഡ് ക്യൂ വേരിയബിൾ അടങ്ങിയിരിക്കുന്നു. ഇത് java.util.TreeSet-നുള്ള ഒരു അഡാപ്റ്ററായ ഒരു ഇഷ്‌ടാനുസൃത ഒബ്‌ജക്‌റ്റാണ്, ഇത് ട്രേഡ് ചെയ്‌ത ഓഹരികളുടെ അവരോഹണ ക്രമത്തിൽ ഉയർന്ന N ഫലങ്ങൾ ട്രാക്കുചെയ്യാൻ ഉപയോഗിക്കുന്നു.

groupBy, mapValues ​​കോളുകൾ നിങ്ങൾ ഇതിനകം കണ്ടിട്ടുണ്ട്, അതിനാൽ ഞങ്ങൾ അതിലേക്ക് പോകില്ല (KTable.toStream രീതിയെ ഞങ്ങൾ വിളിക്കുന്നത് KTable.print രീതി ഒഴിവാക്കിയതിനാലാണ്). എന്നാൽ നിങ്ങൾ ഇതുവരെ മൊത്തം () ന്റെ KTable പതിപ്പ് കണ്ടിട്ടില്ല, അതിനാൽ ഞങ്ങൾ അത് ചർച്ച ചെയ്യാൻ കുറച്ച് സമയം ചെലവഴിക്കും.

നിങ്ങൾ ഓർക്കുന്നതുപോലെ, ഒരേ കീകളുള്ള റെക്കോർഡുകൾ അപ്‌ഡേറ്റുകളായി കണക്കാക്കുന്നു എന്നതാണ് കെ‌ടേബിളിനെ വ്യത്യസ്തമാക്കുന്നത്. KTable പഴയ എൻട്രിക്ക് പകരം പുതിയത് നൽകുന്നു. അഗ്രഗേഷൻ സമാനമായ രീതിയിൽ സംഭവിക്കുന്നു: ഒരേ കീ ഉള്ള ഏറ്റവും പുതിയ റെക്കോർഡുകൾ സമാഹരിച്ചിരിക്കുന്നു. ഒരു റെക്കോർഡ് വരുമ്പോൾ, അത് ഒരു ആഡർ (മൊത്തം രീതിയിലുള്ള കോളിലെ രണ്ടാമത്തെ പാരാമീറ്റർ) ഉപയോഗിച്ച് FixedSizePriorityQueue ക്ലാസ് ഇൻസ്‌റ്റൻസിലേക്ക് ചേർക്കുന്നു, എന്നാൽ അതേ കീ ഉപയോഗിച്ച് മറ്റൊരു റെക്കോർഡ് ഇതിനകം നിലവിലുണ്ടെങ്കിൽ, ഒരു സബ്‌ട്രാക്ടർ ഉപയോഗിച്ച് പഴയ റെക്കോർഡ് നീക്കംചെയ്യപ്പെടും (ഇതിൽ മൂന്നാമത്തെ പാരാമീറ്റർ മൊത്തം രീതി കോൾ).

ഇതിനർത്ഥം ഞങ്ങളുടെ അഗ്രഗേറ്റർ, FixedSizePriorityQueue, എല്ലാ മൂല്യങ്ങളും ഒരു കീ ഉപയോഗിച്ച് സമാഹരിക്കുന്നില്ല, എന്നാൽ N ഏറ്റവുമധികം ട്രേഡ് ചെയ്യപ്പെടുന്ന സ്റ്റോക്കുകളുടെ അളവുകളുടെ ചലിക്കുന്ന തുക സംഭരിക്കുന്നു എന്നാണ്. ഓരോ ഇൻകമിംഗ് എൻട്രിയിലും ഇതുവരെ വിറ്റഴിച്ച മൊത്തം ഓഹരികളുടെ എണ്ണം അടങ്ങിയിരിക്കുന്നു. ഓരോ അപ്‌ഡേറ്റിന്റെയും റോളിംഗ് അഗ്രഗേഷൻ ആവശ്യമില്ലാതെ, നിലവിൽ ഏതൊക്കെ കമ്പനികളുടെ ഓഹരികളാണ് ഏറ്റവും കൂടുതൽ ട്രേഡ് ചെയ്യപ്പെടുന്നത് എന്നതിനെക്കുറിച്ചുള്ള വിവരങ്ങൾ KTable നിങ്ങൾക്ക് നൽകും.

രണ്ട് പ്രധാന കാര്യങ്ങൾ ചെയ്യാൻ ഞങ്ങൾ പഠിച്ചു:

• ഒരു പൊതു കീ ഉപയോഗിച്ച് KTable-ലെ ഗ്രൂപ്പ് മൂല്യങ്ങൾ;
• ഈ ഗ്രൂപ്പുചെയ്ത മൂല്യങ്ങളിൽ റോളപ്പ്, അഗ്രഗേഷൻ എന്നിവ പോലുള്ള ഉപയോഗപ്രദമായ പ്രവർത്തനങ്ങൾ നടത്തുക.

ഒരു കാഫ്ക സ്ട്രീംസ് ആപ്ലിക്കേഷനിലൂടെ നീങ്ങുന്ന ഡാറ്റയുടെ അർത്ഥം മനസ്സിലാക്കുന്നതിനും അത് ഏത് വിവരമാണ് വഹിക്കുന്നതെന്ന് മനസ്സിലാക്കുന്നതിനും ഈ പ്രവർത്തനങ്ങൾ എങ്ങനെ നിർവഹിക്കണമെന്ന് അറിയുന്നത് പ്രധാനമാണ്.

ഈ പുസ്തകത്തിൽ നേരത്തെ ചർച്ച ചെയ്ത ചില പ്രധാന ആശയങ്ങളും ഞങ്ങൾ ഒരുമിച്ച് കൊണ്ടുവന്നിട്ടുണ്ട്. അദ്ധ്യായം 4-ൽ, ഒരു സ്ട്രീമിംഗ് ആപ്ലിക്കേഷന് തെറ്റ്-സഹിഷ്ണുത, പ്രാദേശിക സംസ്ഥാനം എത്രത്തോളം പ്രധാനമാണെന്ന് ഞങ്ങൾ ചർച്ച ചെയ്തു. ഈ അധ്യായത്തിലെ ആദ്യ ഉദാഹരണം, പ്രാദേശിക സംസ്ഥാനം വളരെ പ്രധാനമായിരിക്കുന്നത് എന്തുകൊണ്ടാണെന്ന് തെളിയിക്കുന്നു-നിങ്ങൾ ഇതിനകം കണ്ട വിവരങ്ങൾ ട്രാക്ക് ചെയ്യാനുള്ള കഴിവ് ഇത് നൽകുന്നു. ലോക്കൽ ആക്‌സസ്സ് നെറ്റ്‌വർക്ക് കാലതാമസം ഒഴിവാക്കുന്നു, ഇത് ആപ്ലിക്കേഷനെ കൂടുതൽ കാര്യക്ഷമവും പിശക്-പ്രതിരോധശേഷിയുള്ളതുമാക്കുന്നു.

ഏതെങ്കിലും റോളപ്പ് അല്ലെങ്കിൽ അഗ്രഗേഷൻ പ്രവർത്തനം നടത്തുമ്പോൾ, നിങ്ങൾ സ്റ്റേറ്റ് സ്റ്റോറിന്റെ പേര് വ്യക്തമാക്കണം. റോളപ്പും അഗ്രഗേഷൻ പ്രവർത്തനങ്ങളും ഒരു KTable ഉദാഹരണം നൽകുന്നു, കൂടാതെ KTable പഴയ ഫലങ്ങൾ പുതിയവ ഉപയോഗിച്ച് മാറ്റിസ്ഥാപിക്കാൻ സ്റ്റേറ്റ് സ്റ്റോറേജ് ഉപയോഗിക്കുന്നു. നിങ്ങൾ കണ്ടതുപോലെ, എല്ലാ അപ്‌ഡേറ്റുകളും പൈപ്പ്‌ലൈനിലേക്ക് അയയ്‌ക്കപ്പെടുന്നില്ല, സംഗ്രഹ വിവരങ്ങൾ നിർമ്മിക്കുന്നതിനാണ് സംഗ്രഹ പ്രവർത്തനങ്ങൾ രൂപകൽപ്പന ചെയ്‌തിരിക്കുന്നതിനാൽ ഇത് പ്രധാനമാണ്. നിങ്ങൾ പ്രാദേശിക സംസ്ഥാനം പ്രയോഗിക്കുന്നില്ലെങ്കിൽ, എല്ലാ സംയോജനവും റോളപ്പ് ഫലങ്ങളും KTable കൈമാറും.

അടുത്തതായി, ഒരു നിശ്ചിത സമയത്തിനുള്ളിൽ അഗ്രഗേഷൻ പോലുള്ള പ്രവർത്തനങ്ങൾ ഞങ്ങൾ നോക്കാം - വിൻഡോയിംഗ് പ്രവർത്തനങ്ങൾ എന്ന് വിളിക്കപ്പെടുന്നവ.

5.3.2. വിൻഡോ പ്രവർത്തനങ്ങൾ

മുമ്പത്തെ വിഭാഗത്തിൽ, ഞങ്ങൾ സ്ലൈഡിംഗ് കൺവ്യൂഷനും അഗ്രഗേഷനും അവതരിപ്പിച്ചു. ആപ്ലിക്കേഷൻ സ്റ്റോക്ക് സെയിൽസ് വോളിയത്തിന്റെ തുടർച്ചയായ റോൾ-അപ്പ് നടത്തി, തുടർന്ന് എക്സ്ചേഞ്ചിൽ ഏറ്റവും കൂടുതൽ ട്രേഡ് ചെയ്യപ്പെടുന്ന അഞ്ച് സ്റ്റോക്കുകളുടെ സമാഹാരം.

ചിലപ്പോൾ അത്തരം തുടർച്ചയായ സമാഹരണവും ഫലങ്ങളുടെ റോൾ-അപ്പും ആവശ്യമാണ്. ചിലപ്പോൾ നിങ്ങൾ ഒരു നിശ്ചിത കാലയളവിൽ മാത്രം പ്രവർത്തനങ്ങൾ നടത്തേണ്ടതുണ്ട്. ഉദാഹരണത്തിന്, കഴിഞ്ഞ 10 മിനിറ്റിനുള്ളിൽ ഒരു പ്രത്യേക കമ്പനിയുടെ ഓഹരികൾ ഉപയോഗിച്ച് എത്ര എക്സ്ചേഞ്ച് ഇടപാടുകൾ നടത്തി എന്ന് കണക്കാക്കുക. അല്ലെങ്കിൽ കഴിഞ്ഞ 15 മിനിറ്റിനുള്ളിൽ എത്ര ഉപയോക്താക്കൾ ഒരു പുതിയ പരസ്യ ബാനറിൽ ക്ലിക്ക് ചെയ്തു. ഒരു ആപ്ലിക്കേഷൻ അത്തരം പ്രവർത്തനങ്ങൾ ഒന്നിലധികം തവണ ചെയ്തേക്കാം, എന്നാൽ നിശ്ചിത സമയങ്ങളിൽ (ടൈം വിൻഡോകൾ) മാത്രം ബാധകമാകുന്ന ഫലങ്ങൾ.

വാങ്ങുന്നയാൾ മുഖേനയുള്ള എക്സ്ചേഞ്ച് ഇടപാടുകൾ കണക്കാക്കുന്നു

അടുത്ത ഉദാഹരണത്തിൽ, ഒന്നിലധികം വ്യാപാരികളിലുടനീളമുള്ള സ്റ്റോക്ക് ഇടപാടുകൾ ഞങ്ങൾ ട്രാക്ക് ചെയ്യും-ഒന്നുകിൽ വലിയ ഓർഗനൈസേഷനുകൾ അല്ലെങ്കിൽ മികച്ച വ്യക്തിഗത ധനകാര്യകർത്താക്കൾ.

ഈ ട്രാക്കിംഗിന് രണ്ട് കാരണങ്ങളുണ്ട്. വിപണിയിലെ നേതാക്കൾ എന്താണ് വാങ്ങുന്നത് / വിൽക്കുന്നത് എന്ന് അറിയേണ്ടത് അതിലൊന്നാണ്. ഈ വലിയ കളിക്കാരും സങ്കീർണ്ണമായ നിക്ഷേപകരും അവസരം കാണുന്നുവെങ്കിൽ, അവരുടെ തന്ത്രം പിന്തുടരുന്നതിൽ അർത്ഥമുണ്ട്. രണ്ടാമത്തെ കാരണം, നിയമവിരുദ്ധമായ ഇൻസൈഡർ ട്രേഡിംഗിന്റെ സാധ്യമായ ഏതെങ്കിലും അടയാളങ്ങൾ കണ്ടെത്താനുള്ള ആഗ്രഹമാണ്. ഇത് ചെയ്യുന്നതിന്, പ്രധാനപ്പെട്ട പ്രസ് റിലീസുകളുമായുള്ള വലിയ വിൽപ്പന സ്പൈക്കുകളുടെ പരസ്പരബന്ധം നിങ്ങൾ വിശകലനം ചെയ്യേണ്ടതുണ്ട്.

അത്തരം ട്രാക്കിംഗ് ഇനിപ്പറയുന്ന ഘട്ടങ്ങൾ ഉൾക്കൊള്ളുന്നു:

• സ്റ്റോക്ക്-ഇടപാടുകൾ വിഷയത്തിൽ നിന്ന് വായിക്കാൻ ഒരു സ്ട്രീം സൃഷ്ടിക്കുന്നു;
• വാങ്ങുന്നയാൾ ഐഡിയും സ്റ്റോക്ക് ചിഹ്നവും അനുസരിച്ച് ഇൻകമിംഗ് റെക്കോർഡുകൾ ഗ്രൂപ്പുചെയ്യുന്നു. ഗ്രൂപ്പ് ബൈ മെത്തേഡ് എന്ന് വിളിക്കുന്നത് KGroupedStream ക്ലാസ്സിന്റെ ഒരു ഉദാഹരണം നൽകുന്നു;
• KGroupedStream.windowedBy രീതി ഒരു ടൈം വിൻഡോയിലേക്ക് പരിമിതപ്പെടുത്തിയിരിക്കുന്ന ഒരു ഡാറ്റ സ്ട്രീം നൽകുന്നു, ഇത് വിൻഡോഡ് അഗ്രഗേഷൻ അനുവദിക്കുന്നു. വിൻഡോ തരം അനുസരിച്ച്, ഒരു TimeWindowedKStream അല്ലെങ്കിൽ ഒരു SessionWindowedKStream നൽകുന്നു;
• അഗ്രഗേഷൻ പ്രവർത്തനത്തിനുള്ള ഇടപാടുകളുടെ എണ്ണം. ഈ കണക്കിൽ ഒരു പ്രത്യേക റെക്കോർഡ് കണക്കിലെടുക്കുന്നുണ്ടോ എന്ന് വിൻഡോഡ് ഡാറ്റാ ഫ്ലോ നിർണ്ണയിക്കുന്നു;
• വികസന സമയത്ത് ഒരു വിഷയത്തിലേക്ക് ഫലങ്ങൾ എഴുതുകയോ കൺസോളിലേക്ക് ഔട്ട്പുട്ട് ചെയ്യുകയോ ചെയ്യുക.

ഈ ആപ്ലിക്കേഷന്റെ ടോപ്പോളജി ലളിതമാണ്, എന്നാൽ അതിന്റെ വ്യക്തമായ ചിത്രം സഹായകമാകും. നമുക്ക് ചിത്രം നോക്കാം. 5.11

അടുത്തതായി, വിൻഡോ പ്രവർത്തനങ്ങളുടെ പ്രവർത്തനക്ഷമതയും അനുബന്ധ കോഡും ഞങ്ങൾ നോക്കാം.

വിൻഡോ തരങ്ങൾ

കാഫ്ക സ്ട്രീമുകളിൽ മൂന്ന് തരം വിൻഡോകൾ ഉണ്ട്:

• സെഷനൽ;
• "ടമ്പിംഗ്" (ടമ്പിംഗ്);
• സ്ലൈഡിംഗ്/ഹോപ്പിംഗ്.

ഏതാണ് തിരഞ്ഞെടുക്കേണ്ടത് എന്നത് നിങ്ങളുടെ ബിസിനസ്സ് ആവശ്യകതകളെ ആശ്രയിച്ചിരിക്കുന്നു. ടംബ്ലിംഗ്, ജമ്പിംഗ് വിൻഡോകൾ സമയ പരിമിതമാണ്, അതേസമയം സെഷൻ വിൻഡോകൾ ഉപയോക്തൃ പ്രവർത്തനത്താൽ പരിമിതപ്പെടുത്തിയിരിക്കുന്നു-സെഷന്റെ (ങ്ങളുടെ) ദൈർഘ്യം നിർണ്ണയിക്കുന്നത് ഉപയോക്താവ് എത്രത്തോളം സജീവമാണ് എന്നതിനെ അടിസ്ഥാനമാക്കിയാണ്. എല്ലാ വിൻഡോ തരങ്ങളും എൻട്രികളുടെ തീയതി/സമയ സ്റ്റാമ്പുകളെ അടിസ്ഥാനമാക്കിയുള്ളതാണ്, സിസ്റ്റം സമയമല്ല എന്നതാണ് ഓർമ്മിക്കേണ്ട പ്രധാന കാര്യം.

അടുത്തതായി, ഓരോ വിൻഡോ തരത്തിലും ഞങ്ങൾ ടോപ്പോളജി നടപ്പിലാക്കുന്നു. പൂർണ്ണമായ കോഡ് ആദ്യ ഉദാഹരണത്തിൽ മാത്രമേ നൽകൂ; മറ്റ് തരത്തിലുള്ള വിൻഡോകൾക്ക് വിൻഡോ ഓപ്പറേഷൻ തരം ഒഴികെ മറ്റൊന്നും മാറില്ല.

സെഷൻ വിൻഡോകൾ

സെഷൻ വിൻഡോകൾ മറ്റെല്ലാ തരത്തിലുള്ള വിൻഡോകളിൽ നിന്നും വളരെ വ്യത്യസ്തമാണ്. ഉപയോക്താവിന്റെ (അല്ലെങ്കിൽ നിങ്ങൾ ട്രാക്ക് ചെയ്യാൻ ആഗ്രഹിക്കുന്ന എന്റിറ്റിയുടെ പ്രവർത്തനം) സമയത്തിനനുസരിച്ച് അവ പരിമിതപ്പെടുത്തിയിട്ടില്ല. സെഷൻ വിൻഡോകൾ നിഷ്‌ക്രിയത്വത്തിന്റെ കാലഘട്ടങ്ങളാൽ വേർതിരിച്ചിരിക്കുന്നു.

ചിത്രം 5.12 സെഷൻ വിൻഡോകളുടെ ആശയം വ്യക്തമാക്കുന്നു. ചെറിയ സെഷൻ അതിന്റെ ഇടതുവശത്തുള്ള സെഷനുമായി ലയിക്കും. വലത് വശത്തുള്ള സെഷൻ വെവ്വേറെയായിരിക്കും, കാരണം അത് ദീർഘനാളത്തെ നിഷ്ക്രിയത്വത്തെ പിന്തുടരുന്നു. സെഷൻ വിൻഡോകൾ ഉപയോക്തൃ പ്രവർത്തനത്തെ അടിസ്ഥാനമാക്കിയുള്ളതാണ്, എന്നാൽ എൻട്രി ഏത് സെഷനിൽ ഉൾപ്പെട്ടതാണെന്ന് നിർണ്ണയിക്കാൻ എൻട്രികളിൽ നിന്നുള്ള തീയതി/സമയ സ്റ്റാമ്പുകൾ ഉപയോഗിക്കുക.

സ്റ്റോക്ക് ഇടപാടുകൾ ട്രാക്ക് ചെയ്യുന്നതിന് സെഷൻ വിൻഡോകൾ ഉപയോഗിക്കുന്നു

എക്സ്ചേഞ്ച് ഇടപാടുകളെക്കുറിച്ചുള്ള വിവരങ്ങൾ ക്യാപ്‌ചർ ചെയ്യാൻ നമുക്ക് സെഷൻ വിൻഡോകൾ ഉപയോഗിക്കാം. സെഷൻ വിൻഡോകൾ നടപ്പിലാക്കുന്നത് ലിസ്റ്റിംഗ് 5.5-ൽ കാണിച്ചിരിക്കുന്നു (ഇത് src/main/java/bbejeck/chapter_5/CountingWindowingAndKTableJoinExample.java-ൽ കാണാം).

ഈ ടോപ്പോളജിയിലെ ഒട്ടുമിക്ക പ്രവർത്തനങ്ങളും നിങ്ങൾ ഇതിനകം കണ്ടിട്ടുണ്ട്, അതിനാൽ അവ ഇവിടെ വീണ്ടും നോക്കേണ്ടതില്ല. എന്നാൽ ഇവിടെ നിരവധി പുതിയ ഘടകങ്ങളും ഉണ്ട്, അത് നമ്മൾ ഇപ്പോൾ ചർച്ച ചെയ്യും.

ഏതൊരു ഗ്രൂപ്പ് ബൈ ഓപ്പറേഷനും സാധാരണയായി ഏതെങ്കിലും തരത്തിലുള്ള അഗ്രഗേഷൻ ഓപ്പറേഷൻ നടത്തുന്നു (അഗ്രഗേഷൻ, റോളപ്പ് അല്ലെങ്കിൽ കൗണ്ടിംഗ്). പ്രവർത്തിക്കുന്ന ആകെത്തുക ഉപയോഗിച്ച് നിങ്ങൾക്ക് ക്യുമുലേറ്റീവ് അഗ്രഗേഷൻ നടത്താം, അല്ലെങ്കിൽ ഒരു നിശ്ചിത സമയ വിൻഡോയ്ക്കുള്ളിൽ റെക്കോർഡുകൾ കണക്കിലെടുക്കുന്ന വിൻഡോ അഗ്രഗേഷൻ നടത്താം.

ലിസ്റ്റിംഗ് 5.5 ലെ കോഡ് സെഷൻ വിൻഡോകളിലെ ഇടപാടുകളുടെ എണ്ണം കണക്കാക്കുന്നു. ചിത്രത്തിൽ. 5.13 ഈ പ്രവർത്തനങ്ങൾ ഘട്ടം ഘട്ടമായി വിശകലനം ചെയ്യുന്നു.

windowedBy(SessionWindows.with(twentySeconds) വരെ (പതിനഞ്ച് മിനിറ്റ്)) 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 നും.

ഈ വിഭാഗത്തിൽ വിവരിച്ചതിൽ നിന്ന്, ഇനിപ്പറയുന്ന പ്രധാന സൂക്ഷ്മതകൾ ഓർമ്മിക്കുന്നത് മൂല്യവത്താണ്:

• സെഷനുകൾ നിശ്ചിത വലിപ്പത്തിലുള്ള വിൻഡോകളല്ല. ഒരു സെഷന്റെ ദൈർഘ്യം നിർണ്ണയിക്കുന്നത് ഒരു നിശ്ചിത കാലയളവിനുള്ളിലെ പ്രവർത്തനമാണ്;
• ഡാറ്റയിലെ തീയതി/സമയ സ്റ്റാമ്പുകൾ ഇവന്റ് നിലവിലുള്ള ഒരു സെഷനിൽ ആണോ അതോ നിഷ്‌ക്രിയ കാലയളവിലാണോ എന്ന് നിർണ്ണയിക്കുന്നു.

അടുത്തതായി നമ്മൾ അടുത്ത തരം വിൻഡോ ചർച്ച ചെയ്യും - "ടമ്പിംഗ്" വിൻഡോകൾ.

"ടമ്പിംഗ്" വിൻഡോകൾ

ഒരു നിശ്ചിത കാലയളവിനുള്ളിൽ സംഭവിക്കുന്ന ഇവന്റുകൾ ടമ്പിംഗ് വിൻഡോകൾ ക്യാപ്‌ചർ ചെയ്യുന്നു. ഓരോ 20 സെക്കൻഡിലും ഒരു നിശ്ചിത കമ്പനിയുടെ എല്ലാ സ്റ്റോക്ക് ഇടപാടുകളും നിങ്ങൾ പിടിച്ചെടുക്കേണ്ടതുണ്ടെന്ന് സങ്കൽപ്പിക്കുക, അതിനാൽ ആ കാലയളവിൽ നിങ്ങൾ എല്ലാ ഇവന്റുകളും ശേഖരിക്കുന്നു. 20-സെക്കൻഡ് ഇടവേളയുടെ അവസാനം, വിൻഡോ ഉരുട്ടി പുതിയ 20-സെക്കൻഡ് നിരീക്ഷണ ഇടവേളയിലേക്ക് നീങ്ങുന്നു. ചിത്രം 5.14 ഈ സാഹചര്യം വ്യക്തമാക്കുന്നു.

നിങ്ങൾക്ക് കാണാനാകുന്നതുപോലെ, കഴിഞ്ഞ 20 സെക്കൻഡിനുള്ളിൽ ലഭിച്ച എല്ലാ ഇവന്റുകളും വിൻഡോയിൽ ഉൾപ്പെടുത്തിയിട്ടുണ്ട്. ഈ കാലയളവിന്റെ അവസാനം, ഒരു പുതിയ വിൻഡോ സൃഷ്ടിക്കപ്പെടുന്നു.

ഓരോ 5.6 സെക്കൻഡിലും സ്റ്റോക്ക് ഇടപാടുകൾ ക്യാപ്‌ചർ ചെയ്യുന്നതിന് ടംബ്ലിംഗ് വിൻഡോകളുടെ ഉപയോഗം കാണിക്കുന്ന കോഡ് ലിസ്റ്റിംഗ് 20 കാണിക്കുന്നു (src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java-ൽ കാണപ്പെടുന്നു).

TimeWindows.of മെത്തേഡ് കോളിലേക്കുള്ള ഈ ചെറിയ മാറ്റം ഉപയോഗിച്ച്, നിങ്ങൾക്ക് ഒരു ടംബ്ലിംഗ് വിൻഡോ ഉപയോഗിക്കാം. ഈ ഉദാഹരണം വരെ () രീതിയെ വിളിക്കുന്നില്ല, അതിനാൽ 24 മണിക്കൂർ സ്ഥിരസ്ഥിതി നിലനിർത്തൽ ഇടവേള ഉപയോഗിക്കും.

അവസാനമായി, വിൻഡോ ഓപ്ഷനുകളുടെ അവസാനത്തിലേക്ക് നീങ്ങാൻ സമയമായി - "ഹോപ്പിംഗ്" വിൻഡോകൾ.

സ്ലൈഡിംഗ് ("ജമ്പിംഗ്") വിൻഡോകൾ

സ്ലൈഡിംഗ്/ഹോപ്പിംഗ് വിൻഡോകൾ ടംബ്ലിംഗ് വിൻഡോകൾക്ക് സമാനമാണ്, എന്നാൽ ചെറിയ വ്യത്യാസമുണ്ട്. സമീപകാല ഇവന്റുകൾ പ്രോസസ്സ് ചെയ്യുന്നതിന് ഒരു പുതിയ വിൻഡോ സൃഷ്ടിക്കുന്നതിന് മുമ്പ് സ്ലൈഡിംഗ് വിൻഡോകൾ സമയ ഇടവേളയുടെ അവസാനം വരെ കാത്തിരിക്കില്ല. വിൻഡോ ദൈർഘ്യത്തേക്കാൾ കുറഞ്ഞ കാത്തിരിപ്പ് ഇടവേളയ്ക്ക് ശേഷം അവർ പുതിയ കണക്കുകൂട്ടലുകൾ ആരംഭിക്കുന്നു.

വിൻഡോകൾ ഇടിക്കുന്നതും ചാടുന്നതും തമ്മിലുള്ള വ്യത്യാസം വ്യക്തമാക്കുന്നതിന്, സ്റ്റോക്ക് എക്‌സ്‌ചേഞ്ച് ഇടപാടുകൾ കണക്കാക്കുന്നതിനുള്ള ഉദാഹരണത്തിലേക്ക് നമുക്ക് മടങ്ങാം. ഇടപാടുകളുടെ എണ്ണം കണക്കാക്കുക എന്നതാണ് ഞങ്ങളുടെ ലക്ഷ്യം, എന്നാൽ കൗണ്ടർ അപ്‌ഡേറ്റ് ചെയ്യുന്നതിന് മുമ്പ് മുഴുവൻ സമയവും കാത്തിരിക്കാൻ ഞങ്ങൾ ആഗ്രഹിക്കുന്നില്ല. പകരം, ഞങ്ങൾ ചെറിയ ഇടവേളകളിൽ കൗണ്ടർ അപ്ഡേറ്റ് ചെയ്യും. ഉദാഹരണത്തിന്, ഞങ്ങൾ ഇപ്പോഴും ഓരോ 20 സെക്കൻഡിലും ഇടപാടുകളുടെ എണ്ണം കണക്കാക്കും, എന്നാൽ ചിത്രത്തിൽ കാണിച്ചിരിക്കുന്നതുപോലെ ഓരോ 5 സെക്കൻഡിലും കൗണ്ടർ അപ്ഡേറ്റ് ചെയ്യും. 5.15 ഈ സാഹചര്യത്തിൽ, ഡാറ്റ ഓവർലാപ്പുചെയ്യുന്ന മൂന്ന് ഫല വിൻഡോകളിൽ ഞങ്ങൾ അവസാനിക്കും.

സ്ലൈഡിംഗ് വിൻഡോകൾ നിർവചിക്കുന്നതിനുള്ള കോഡ് ലിസ്റ്റിംഗ് 5.7 കാണിക്കുന്നു (src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java-ൽ കാണപ്പെടുന്നു).

അഡ്വാൻസ്‌ബൈ() രീതിയിലേക്ക് ഒരു കോൾ ചേർത്ത് ഒരു ടംബ്ലിംഗ് വിൻഡോ ഹോപ്പിംഗ് വിൻഡോയിലേക്ക് പരിവർത്തനം ചെയ്യാവുന്നതാണ്. കാണിച്ചിരിക്കുന്ന ഉദാഹരണത്തിൽ, സേവിംഗ് ഇടവേള 15 മിനിറ്റാണ്.

അഗ്രഗേഷൻ ഫലങ്ങൾ ടൈം വിൻഡോകളിലേക്ക് പരിമിതപ്പെടുത്തുന്നത് എങ്ങനെയെന്ന് നിങ്ങൾ ഈ വിഭാഗത്തിൽ കണ്ടു. പ്രത്യേകിച്ചും, ഈ വിഭാഗത്തിൽ നിന്ന് ഇനിപ്പറയുന്ന മൂന്ന് കാര്യങ്ങൾ നിങ്ങൾ ഓർക്കണമെന്ന് ഞാൻ ആഗ്രഹിക്കുന്നു:

• സെഷൻ വിൻഡോകളുടെ വലുപ്പം പരിമിതപ്പെടുത്തിയിരിക്കുന്നത് സമയ കാലയളവിനല്ല, ഉപയോക്തൃ പ്രവർത്തനത്തിലൂടെയാണ്;
• "ടംബിംഗ്" വിൻഡോകൾ ഒരു നിശ്ചിത സമയത്തിനുള്ളിൽ ഇവന്റുകളുടെ ഒരു അവലോകനം നൽകുന്നു;
• ജമ്പിംഗ് വിൻഡോകളുടെ ദൈർഘ്യം നിശ്ചയിച്ചിട്ടുണ്ട്, എന്നാൽ അവ പതിവായി അപ്ഡേറ്റ് ചെയ്യപ്പെടുന്നു, കൂടാതെ എല്ലാ വിൻഡോകളിലും ഓവർലാപ്പിംഗ് എൻട്രികൾ അടങ്ങിയിരിക്കാം.

അടുത്തതായി, ഒരു കണക്ഷനായി KTable ഒരു KStream-ലേക്ക് എങ്ങനെ പരിവർത്തനം ചെയ്യാമെന്ന് നമ്മൾ പഠിക്കും.

5.3.3. KStream, KTable ഒബ്‌ജക്‌റ്റുകൾ ബന്ധിപ്പിക്കുന്നു

അദ്ധ്യായം 4-ൽ, രണ്ട് KStream ഒബ്‌ജക്റ്റുകളെ ബന്ധിപ്പിക്കുന്നതിനെക്കുറിച്ച് ഞങ്ങൾ ചർച്ച ചെയ്തു. ഇനി KTable ഉം KStream ഉം എങ്ങനെ കണക്ട് ചെയ്യാം എന്ന് പഠിക്കണം. ഇനിപ്പറയുന്ന ലളിതമായ കാരണത്താൽ ഇത് ആവശ്യമായി വന്നേക്കാം. KStream എന്നത് റെക്കോർഡുകളുടെ ഒരു സ്ട്രീം ആണ്, കൂടാതെ KTable എന്നത് റെക്കോർഡ് അപ്ഡേറ്റുകളുടെ ഒരു സ്ട്രീം ആണ്, എന്നാൽ ചിലപ്പോൾ നിങ്ങൾക്ക് KTable-ൽ നിന്നുള്ള അപ്ഡേറ്റുകൾ ഉപയോഗിച്ച് റെക്കോർഡ് സ്ട്രീമിലേക്ക് അധിക സന്ദർഭം ചേർക്കാൻ താൽപ്പര്യപ്പെട്ടേക്കാം.

സ്റ്റോക്ക് എക്‌സ്‌ചേഞ്ച് ഇടപാടുകളുടെ എണ്ണത്തെക്കുറിച്ചുള്ള ഡാറ്റ എടുത്ത് അവയെ പ്രസക്തമായ വ്യവസായങ്ങൾക്കായുള്ള സ്റ്റോക്ക് എക്‌സ്‌ചേഞ്ച് വാർത്തകളുമായി സംയോജിപ്പിക്കാം. നിങ്ങൾക്ക് ഇതിനകം ഉള്ള കോഡ് നൽകി ഇത് നേടുന്നതിന് നിങ്ങൾ ചെയ്യേണ്ടത് ഇതാ.

1. സ്റ്റോക്ക് ഇടപാടുകളുടെ എണ്ണത്തെക്കുറിച്ചുള്ള ഡാറ്റയുള്ള ഒരു KTable ഒബ്‌ജക്റ്റ് ഒരു KStream ആക്കി മാറ്റുക, തുടർന്ന് ഈ സ്റ്റോക്ക് ചിഹ്നവുമായി ബന്ധപ്പെട്ട വ്യവസായ മേഖലയെ സൂചിപ്പിക്കുന്ന കീ ഉപയോഗിച്ച് കീ മാറ്റിസ്ഥാപിക്കുക.
2. സ്റ്റോക്ക് എക്സ്ചേഞ്ച് വാർത്തകൾക്കൊപ്പം ഒരു വിഷയത്തിൽ നിന്നുള്ള ഡാറ്റ വായിക്കുന്ന ഒരു KTable ഒബ്ജക്റ്റ് സൃഷ്ടിക്കുക. ഈ പുതിയ കെ.ടേബിളിനെ വ്യവസായ മേഖലയുടെ അടിസ്ഥാനത്തിൽ തരംതിരിക്കും.
3. വ്യവസായ മേഖലയിലെ സ്റ്റോക്ക് എക്സ്ചേഞ്ച് ഇടപാടുകളുടെ എണ്ണവുമായി വാർത്താ അപ്ഡേറ്റുകൾ ബന്ധിപ്പിക്കുക.

ഇനി ഈ ആക്ഷൻ പ്ലാൻ എങ്ങനെ നടപ്പിലാക്കാം എന്ന് നോക്കാം.

KTable-ലേക്ക് KStream-ലേക്ക് പരിവർത്തനം ചെയ്യുക

KTable-ലേക്ക് KStream-ലേക്ക് പരിവർത്തനം ചെയ്യുന്നതിന് നിങ്ങൾ ഇനിപ്പറയുന്നവ ചെയ്യേണ്ടതുണ്ട്.

1. KTable.toStream() രീതിയിലേക്ക് വിളിക്കുക.
2. KStream.map രീതി വിളിക്കുന്നതിലൂടെ, വ്യവസായത്തിന്റെ പേര് ഉപയോഗിച്ച് കീ മാറ്റിസ്ഥാപിക്കുക, തുടർന്ന് വിൻ‌ഡോഡ് ഇൻസ്‌റ്റൻസിൽ നിന്ന് ഇടപാട് സംഗ്രഹം ഒബ്‌ജക്റ്റ് വീണ്ടെടുക്കുക.

ഞങ്ങൾ ഈ പ്രവർത്തനങ്ങളെ ഇനിപ്പറയുന്ന രീതിയിൽ ഒരുമിച്ച് ബന്ധിപ്പിക്കും (കോഡ് src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java എന്ന ഫയലിൽ കാണാം) (ലിസ്റ്റിംഗ് 5.8).

ഞങ്ങൾ ഒരു KStream.map ഓപ്പറേഷൻ നടത്തുന്നതിനാൽ, ഒരു കണക്ഷനിൽ ഉപയോഗിക്കുമ്പോൾ തിരികെ നൽകിയ KStream ഇൻസ്‌റ്റൻസ് യാന്ത്രികമായി വീണ്ടും പാർട്ടീഷൻ ചെയ്യുന്നു.

ഞങ്ങൾ പരിവർത്തന പ്രക്രിയ പൂർത്തിയാക്കി, അടുത്തതായി സ്റ്റോക്ക് വാർത്തകൾ വായിക്കുന്നതിനായി ഒരു KTable ഒബ്‌ജക്റ്റ് സൃഷ്‌ടിക്കേണ്ടതുണ്ട്.

സ്റ്റോക്ക് വാർത്തകൾക്കായി കെ ടേബിളിന്റെ സൃഷ്ടി

ഭാഗ്യവശാൽ, ഒരു കെ‌ടേബിൾ ഒബ്‌ജക്‌റ്റ് സൃഷ്‌ടിക്കുന്നതിന് ഒരു കോഡ് മാത്രമേ എടുക്കൂ (കോഡ് src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java-ൽ കാണാം) (ലിസ്റ്റിംഗ് 5.9).

ക്രമീകരണങ്ങളിൽ സ്ട്രിംഗ് സെർഡെസ് ഉപയോഗിക്കുന്നതിനാൽ, സെർഡെ ഒബ്‌ജക്റ്റുകളൊന്നും വ്യക്തമാക്കേണ്ടതില്ല എന്നത് ശ്രദ്ധിക്കേണ്ടതാണ്. കൂടാതെ, ആദ്യകാല എണ്ണൽ ഉപയോഗിക്കുന്നതിലൂടെ, പട്ടിക തുടക്കത്തിൽ തന്നെ റെക്കോർഡുകൾ കൊണ്ട് നിറഞ്ഞിരിക്കുന്നു.

ഇപ്പോൾ നമുക്ക് അവസാന ഘട്ടത്തിലേക്ക് പോകാം - കണക്ഷൻ.

ഇടപാടുകളുടെ എണ്ണം ഡാറ്റയുമായി വാർത്താ അപ്‌ഡേറ്റുകൾ ബന്ധിപ്പിക്കുന്നു

ഒരു കണക്ഷൻ സൃഷ്ടിക്കുന്നത് ബുദ്ധിമുട്ടുള്ള കാര്യമല്ല. പ്രസക്തമായ വ്യവസായത്തിന് സ്റ്റോക്ക് വാർത്തകൾ ഇല്ലെങ്കിൽ ഞങ്ങൾ ഇടത് ചേരൽ ഉപയോഗിക്കും (ആവശ്യമായ കോഡ് ഫയലിൽ src/main/java/bbejeck/chapter_5/CountingWindowingAndKtableJoinExample.java എന്ന ഫയലിൽ കാണാം) (ലിസ്റ്റിംഗ് 5.10).

ഈ ലെഫ്റ്റ് ജോയിൻ ഓപ്പറേറ്റർ വളരെ ലളിതമാണ്. അധ്യായം 4-ലെ ജോയിനുകളിൽ നിന്ന് വ്യത്യസ്തമായി, JoinWindow രീതി ഉപയോഗിക്കുന്നില്ല, കാരണം ഒരു KStream-KTable ജോയിൻ ചെയ്യുമ്പോൾ, ഓരോ കീയ്ക്കും KTable-ൽ ഒരു എൻട്രി മാത്രമേയുള്ളൂ. അത്തരമൊരു കണക്ഷൻ സമയബന്ധിതമായി പരിമിതപ്പെടുത്തിയിട്ടില്ല: റെക്കോർഡ് ഒന്നുകിൽ കെ.ടേബിളിൽ അല്ലെങ്കിൽ ഇല്ല. പ്രധാന നിഗമനം: KTable ഒബ്‌ജക്‌റ്റുകൾ ഉപയോഗിച്ച് നിങ്ങൾക്ക് പതിവായി അപ്‌ഡേറ്റ് ചെയ്യുന്ന റഫറൻസ് ഡാറ്റ ഉപയോഗിച്ച് KStream സമ്പന്നമാക്കാം.

KStream-ൽ നിന്നുള്ള ഇവന്റുകൾ സമ്പന്നമാക്കുന്നതിനുള്ള കൂടുതൽ കാര്യക്ഷമമായ മാർഗം ഞങ്ങൾ നോക്കാം.

5.3.4. GlobalKTable വസ്തുക്കൾ

നിങ്ങൾക്ക് കാണാനാകുന്നതുപോലെ, ഇവന്റ് സ്ട്രീമുകൾ സമ്പുഷ്ടമാക്കുകയോ അവയിലേക്ക് സന്ദർഭം ചേർക്കുകയോ ചെയ്യേണ്ടത് ആവശ്യമാണ്. അധ്യായം 4-ൽ നിങ്ങൾ രണ്ട് KStream ഒബ്‌ജക്റ്റുകൾ തമ്മിലുള്ള കണക്ഷനുകൾ കണ്ടു, മുമ്പത്തെ വിഭാഗത്തിൽ ഒരു KStream-ഉം KTable-ഉം തമ്മിലുള്ള കണക്ഷൻ നിങ്ങൾ കണ്ടു. ഈ സാഹചര്യങ്ങളിലെല്ലാം, ഒരു പുതിയ തരം അല്ലെങ്കിൽ മൂല്യത്തിലേക്ക് കീകൾ മാപ്പ് ചെയ്യുമ്പോൾ ഡാറ്റ സ്ട്രീം വീണ്ടും പാർട്ടീഷൻ ചെയ്യേണ്ടത് ആവശ്യമാണ്. ചിലപ്പോൾ പുനർവിഭജനം വ്യക്തമായും ചിലപ്പോൾ കാഫ്ക സ്ട്രീംസ് അത് യാന്ത്രികമായും ചെയ്യുന്നു. കീകൾ മാറിയതിനാൽ വീണ്ടും വിഭജനം ആവശ്യമാണ്, കൂടാതെ റെക്കോർഡുകൾ പുതിയ വിഭാഗങ്ങളിൽ അവസാനിക്കണം, അല്ലാത്തപക്ഷം കണക്ഷൻ അസാധ്യമായിരിക്കും (ഇത് അദ്ധ്യായം 4 ൽ, ഉപവിഭാഗം 4.2.4 ലെ "ഡാറ്റ വീണ്ടും വിഭജിക്കുന്നു" എന്ന വിഭാഗത്തിൽ ചർച്ചചെയ്തു).

വീണ്ടും വിഭജിക്കുന്നതിന് ചിലവുണ്ട്

വീണ്ടും വിഭജിക്കുന്നതിന് ചിലവുകൾ ആവശ്യമാണ് - ഇന്റർമീഡിയറ്റ് വിഷയങ്ങൾ സൃഷ്ടിക്കുന്നതിനും മറ്റൊരു വിഷയത്തിൽ തനിപ്പകർപ്പ് ഡാറ്റ സംഭരിക്കുന്നതിനുമുള്ള അധിക വിഭവ ചെലവുകൾ; ഈ വിഷയത്തിൽ നിന്നുള്ള എഴുത്തും വായനയും കാരണം വർദ്ധിച്ചുവരുന്ന കാലതാമസം എന്നർത്ഥം. കൂടാതെ, നിങ്ങൾക്ക് ഒന്നിലധികം വശങ്ങളിലോ അളവുകളിലോ ചേരണമെങ്കിൽ, നിങ്ങൾ ജോയിനുകൾ ചെയിൻ ചെയ്യണം, പുതിയ കീകൾ ഉപയോഗിച്ച് റെക്കോർഡുകൾ മാപ്പ് ചെയ്യണം, വീണ്ടും പാർട്ടീഷനിംഗ് പ്രക്രിയ വീണ്ടും പ്രവർത്തിപ്പിക്കുക.

ചെറിയ ഡാറ്റാസെറ്റുകളിലേക്ക് കണക്റ്റുചെയ്യുന്നു

ചില സന്ദർഭങ്ങളിൽ, ബന്ധിപ്പിക്കേണ്ട റഫറൻസ് ഡാറ്റയുടെ അളവ് താരതമ്യേന ചെറുതാണ്, അതിനാൽ അതിന്റെ പൂർണ്ണമായ പകർപ്പുകൾ ഓരോ നോഡിലും എളുപ്പത്തിൽ ഉൾക്കൊള്ളാൻ കഴിയും. ഇതുപോലുള്ള സാഹചര്യങ്ങൾക്ക്, കാഫ്ക സ്ട്രീംസ് GlobalKTable ക്ലാസ് നൽകുന്നു.

GlobalKTable സംഭവങ്ങൾ അദ്വിതീയമാണ്, കാരണം ആപ്ലിക്കേഷൻ എല്ലാ ഡാറ്റയും ഓരോ നോഡുകളിലേക്കും പകർത്തുന്നു. എല്ലാ ഡാറ്റയും ഓരോ നോഡിലും ഉള്ളതിനാൽ, റഫറൻസ് ഡാറ്റ കീ ഉപയോഗിച്ച് ഇവന്റ് സ്ട്രീം പാർട്ടീഷൻ ചെയ്യേണ്ട ആവശ്യമില്ല, അങ്ങനെ അത് എല്ലാ പാർട്ടീഷനുകളിലും ലഭ്യമാകും. GlobalKTable ഒബ്‌ജക്‌റ്റുകൾ ഉപയോഗിച്ച് നിങ്ങൾക്ക് കീലെസ് ജോയിനുകൾ നടത്താനും കഴിയും. ഈ സവിശേഷത പ്രകടമാക്കുന്നതിന് മുമ്പത്തെ ഉദാഹരണങ്ങളിൽ ഒന്നിലേക്ക് മടങ്ങാം.

KStream ഒബ്‌ജക്‌റ്റുകൾ GlobalKTable ഒബ്‌ജക്‌റ്റുകളിലേക്ക് ബന്ധിപ്പിക്കുന്നു

ഉപവിഭാഗം 5.3.2-ൽ, വാങ്ങുന്നവർ വഴിയുള്ള എക്‌സ്‌ചേഞ്ച് ഇടപാടുകളുടെ വിൻഡോ അഗ്രഗേഷൻ ഞങ്ങൾ നടത്തി. ഈ സംഗ്രഹത്തിന്റെ ഫലങ്ങൾ ഇതുപോലെ കാണപ്പെടുന്നു:

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

ഈ ഫലങ്ങൾ ഉദ്ദേശ്യം നിറവേറ്റുന്നുണ്ടെങ്കിലും, ഉപഭോക്താവിന്റെ പേരും കമ്പനിയുടെ മുഴുവൻ പേരും പ്രദർശിപ്പിച്ചിരുന്നെങ്കിൽ അത് കൂടുതൽ ഉപയോഗപ്രദമാകുമായിരുന്നു. ഉപഭോക്തൃ നാമവും കമ്പനിയുടെ പേരും ചേർക്കുന്നതിന്, നിങ്ങൾക്ക് സാധാരണ ജോയിൻ ചെയ്യാവുന്നതാണ്, എന്നാൽ നിങ്ങൾ രണ്ട് കീ മാപ്പിംഗുകളും റീ-പാർട്ടീഷനിംഗും ചെയ്യേണ്ടതുണ്ട്. GlobalKTable ഉപയോഗിച്ച് നിങ്ങൾക്ക് അത്തരം പ്രവർത്തനങ്ങളുടെ ചിലവ് ഒഴിവാക്കാം.

ഇത് ചെയ്യുന്നതിന്, ഞങ്ങൾ ലിസ്റ്റിംഗ് 5.11-ൽ നിന്നുള്ള കൗണ്ട്‌സ്ട്രീം ഒബ്‌ജക്റ്റ് ഉപയോഗിക്കും (അനുബന്ധ കോഡ് src/main/java/bbejeck/chapter_5/GlobalKTableExample.java-ൽ കാണാം) കൂടാതെ രണ്ട് GlobalKTable ഒബ്‌ജക്‌റ്റുകളിലേക്ക് അതിനെ ബന്ധിപ്പിക്കുകയും ചെയ്യും.

ഞങ്ങൾ ഇത് മുമ്പ് ചർച്ച ചെയ്തിട്ടുണ്ട്, അതിനാൽ ഞാൻ ഇത് ആവർത്തിക്കില്ല. എന്നാൽ toStream().map ഫംഗ്‌ഷനിലെ കോഡ് വായനാക്ഷമതയ്‌ക്കായി ഇൻലൈൻ ലാംഡ എക്‌സ്‌പ്രഷനുപകരം ഒരു ഫംഗ്‌ഷൻ ഒബ്‌ജക്‌റ്റിലേക്ക് സംഗ്രഹിച്ചിരിക്കുന്നുവെന്ന് ഞാൻ ശ്രദ്ധിക്കുന്നു.

GlobalKTable-ന്റെ രണ്ട് സംഭവങ്ങൾ പ്രഖ്യാപിക്കുക എന്നതാണ് അടുത്ത ഘട്ടം (കാണിച്ചിരിക്കുന്ന കോഡ് src/main/java/bbejeck/chapter_5/GlobalKTableExample.java എന്ന ഫയലിൽ കാണാം) (ലിസ്റ്റിംഗ് 5.12).

എണ്ണപ്പെട്ട തരങ്ങൾ ഉപയോഗിച്ചാണ് വിഷയങ്ങളുടെ പേരുകൾ വിവരിച്ചിരിക്കുന്നത് എന്നത് ശ്രദ്ധിക്കുക.

ഇപ്പോൾ എല്ലാ ഘടകങ്ങളും തയ്യാറായിക്കഴിഞ്ഞു, കണക്ഷനുള്ള കോഡ് എഴുതുക മാത്രമാണ് അവശേഷിക്കുന്നത് (അത് src/main/java/bbejeck/chapter_5/GlobalKTableExample.java എന്ന ഫയലിൽ കാണാം) (ലിസ്റ്റിംഗ് 5.13).

ഈ കോഡിൽ രണ്ട് ചേരലുകൾ ഉണ്ടെങ്കിലും, അവയുടെ ഫലങ്ങളൊന്നും വെവ്വേറെ ഉപയോഗിക്കാത്തതിനാൽ അവ ചങ്ങലയിൽ ബന്ധിപ്പിച്ചിരിക്കുന്നു. മുഴുവൻ പ്രവർത്തനത്തിന്റെയും അവസാനം ഫലങ്ങൾ പ്രദർശിപ്പിക്കും.

മുകളിലുള്ള ജോയിൻ ഓപ്പറേഷൻ പ്രവർത്തിപ്പിക്കുമ്പോൾ, നിങ്ങൾക്ക് ഇതുപോലുള്ള ഫലങ്ങൾ ലഭിക്കും:

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

സാരാംശം മാറിയിട്ടില്ല, എന്നാൽ ഈ ഫലങ്ങൾ കൂടുതൽ വ്യക്തമാണ്.

നിങ്ങൾ അധ്യായം 4 വരെ കണക്കാക്കുകയാണെങ്കിൽ, നിങ്ങൾ ഇതിനകം തന്നെ നിരവധി തരത്തിലുള്ള കണക്ഷനുകൾ പ്രവർത്തനത്തിൽ കണ്ടിട്ടുണ്ട്. അവ പട്ടികയിൽ പട്ടികപ്പെടുത്തിയിട്ടുണ്ട്. 5.2 ഈ പട്ടിക കാഫ്ക സ്ട്രീമുകളുടെ പതിപ്പ് 1.0.0-ലെ കണക്റ്റിവിറ്റി കഴിവുകളെ പ്രതിഫലിപ്പിക്കുന്നു; ഭാവി റിലീസുകളിൽ എന്തെങ്കിലും മാറ്റമുണ്ടായേക്കാം.

കാര്യങ്ങൾ അവസാനിപ്പിക്കാൻ, നമുക്ക് അടിസ്ഥാനകാര്യങ്ങൾ വീണ്ടും മനസ്സിലാക്കാം: നിങ്ങൾക്ക് പ്രാദേശിക സംസ്ഥാനം ഉപയോഗിച്ച് ഇവന്റ് സ്ട്രീമുകളും (KStream) അപ്ഡേറ്റ് സ്ട്രീമുകളും (KTable) ബന്ധിപ്പിക്കാം. പകരമായി, റഫറൻസ് ഡാറ്റയുടെ വലുപ്പം വളരെ വലുതല്ലെങ്കിൽ, നിങ്ങൾക്ക് GlobalKTable ഒബ്‌ജക്റ്റ് ഉപയോഗിക്കാം. GlobalKTables എല്ലാ പാർട്ടീഷനുകളും ഓരോ കാഫ്ക സ്ട്രീം ആപ്ലിക്കേഷൻ നോഡിലേക്കും പകർത്തുന്നു, കീ ഏത് പാർട്ടീഷനുമായി പൊരുത്തപ്പെടുന്നു എന്നത് പരിഗണിക്കാതെ തന്നെ എല്ലാ ഡാറ്റയും ലഭ്യമാണെന്ന് ഉറപ്പാക്കുന്നു.

അടുത്തതായി നമ്മൾ കാഫ്ക സ്ട്രീം ഫീച്ചർ കാണും, അതിന് നന്ദി കാഫ്ക വിഷയത്തിൽ നിന്നുള്ള ഡാറ്റ ഉപയോഗിക്കാതെ തന്നെ നമുക്ക് സംസ്ഥാന മാറ്റങ്ങൾ നിരീക്ഷിക്കാൻ കഴിയും.

5.3.5. സംശയാസ്പദമായ അവസ്ഥ

ഞങ്ങൾ ഇതിനകം സംസ്ഥാനം ഉൾപ്പെടുന്ന നിരവധി പ്രവർത്തനങ്ങൾ നടത്തി, എല്ലായ്പ്പോഴും കൺസോളിലേക്ക് ഫലങ്ങൾ ഔട്ട്പുട്ട് ചെയ്യുന്നു (വികസന ആവശ്യങ്ങൾക്കായി) അല്ലെങ്കിൽ അവ ഒരു വിഷയത്തിലേക്ക് എഴുതുക (ഉത്പാദന ആവശ്യങ്ങൾക്കായി). ഒരു വിഷയത്തിലേക്ക് ഫലങ്ങൾ എഴുതുമ്പോൾ, അവ കാണുന്നതിന് നിങ്ങൾ ഒരു കാഫ്ക ഉപഭോക്താവിനെ ഉപയോഗിക്കേണ്ടതുണ്ട്.

ഈ വിഷയങ്ങളിൽ നിന്നുള്ള ഡാറ്റ വായിക്കുന്നത് ഒരു തരം മെറ്റീരിയലൈസ്ഡ് കാഴ്‌ചകളായി കണക്കാക്കാം. ഞങ്ങളുടെ ഉദ്ദേശ്യങ്ങൾക്കായി, വിക്കിപീഡിയയിൽ നിന്നുള്ള ഒരു സാമഗ്രി കാഴ്‌ചയുടെ നിർവചനം ഉപയോഗിക്കാം: “...ഒരു അന്വേഷണത്തിന്റെ ഫലങ്ങൾ ഉൾക്കൊള്ളുന്ന ഒരു ഫിസിക്കൽ ഡാറ്റാബേസ് ഒബ്‌ജക്റ്റ്. ഉദാഹരണത്തിന്, ഇത് റിമോട്ട് ഡാറ്റയുടെ പ്രാദേശിക പകർപ്പ് അല്ലെങ്കിൽ ഒരു പട്ടികയുടെ അല്ലെങ്കിൽ ജോയിൻ ഫലങ്ങളുടെ വരികളുടെയും/അല്ലെങ്കിൽ കോളങ്ങളുടെയും ഉപഗണം അല്ലെങ്കിൽ സമാഹരണത്തിലൂടെ ലഭിച്ച ഒരു സംഗ്രഹ പട്ടിക ആകാം" (https://en.wikipedia.org/wiki /Materialized_view).

കാഫ്ക സ്ട്രീംസ് സംസ്ഥാന സ്റ്റോറുകളിൽ സംവേദനാത്മക അന്വേഷണങ്ങൾ പ്രവർത്തിപ്പിക്കാനും നിങ്ങളെ അനുവദിക്കുന്നു, ഈ വസ്തുതാപരമായ കാഴ്ചകൾ നേരിട്ട് വായിക്കാൻ നിങ്ങളെ അനുവദിക്കുന്നു. സ്റ്റേറ്റ് സ്റ്റോറിലേക്കുള്ള അന്വേഷണം ഒരു വായന-മാത്രം പ്രവർത്തനമാണെന്നത് ശ്രദ്ധിക്കേണ്ടതാണ്. നിങ്ങളുടെ ആപ്ലിക്കേഷൻ ഡാറ്റ പ്രോസസ്സ് ചെയ്യുമ്പോൾ ആകസ്മികമായി സ്ഥിതി അസ്ഥിരമാക്കുന്നതിനെക്കുറിച്ച് നിങ്ങൾ വിഷമിക്കേണ്ടതില്ലെന്ന് ഇത് ഉറപ്പാക്കുന്നു.

സംസ്ഥാന സ്റ്റോറുകളിൽ നേരിട്ട് അന്വേഷിക്കാനുള്ള കഴിവ് പ്രധാനമാണ്. കാഫ്ക ഉപഭോക്താവിൽ നിന്ന് ആദ്യം ഡാറ്റ എടുക്കാതെ തന്നെ നിങ്ങൾക്ക് ഡാഷ്‌ബോർഡ് ആപ്ലിക്കേഷനുകൾ സൃഷ്ടിക്കാൻ കഴിയുമെന്നാണ് ഇതിനർത്ഥം. ഡാറ്റ വീണ്ടും എഴുതേണ്ട ആവശ്യമില്ല എന്നതിനാൽ ഇത് ആപ്ലിക്കേഷന്റെ കാര്യക്ഷമതയും വർദ്ധിപ്പിക്കുന്നു:

• ഡാറ്റയുടെ പ്രാദേശികതയ്ക്ക് നന്ദി, അവ വേഗത്തിൽ ആക്സസ് ചെയ്യാൻ കഴിയും;
• ഡാറ്റയുടെ തനിപ്പകർപ്പ് ഒഴിവാക്കിയിരിക്കുന്നു, കാരണം ഇത് ബാഹ്യ സംഭരണത്തിലേക്ക് എഴുതിയിട്ടില്ല.

നിങ്ങൾ ഓർക്കാൻ ഞാൻ ആഗ്രഹിക്കുന്ന പ്രധാന കാര്യം, നിങ്ങളുടെ ആപ്ലിക്കേഷനിൽ നിന്ന് നിങ്ങൾക്ക് നേരിട്ട് സംസ്ഥാനം അന്വേഷിക്കാൻ കഴിയും എന്നതാണ്. ഇത് നിങ്ങൾക്ക് നൽകുന്ന അവസരങ്ങൾ പറഞ്ഞറിയിക്കാൻ കഴിയില്ല. കാഫ്കയിൽ നിന്നുള്ള ഡാറ്റ ഉപയോഗിക്കുന്നതിനും ആപ്ലിക്കേഷനായി ഒരു ഡാറ്റാബേസിൽ റെക്കോർഡുകൾ സംഭരിക്കുന്നതിനുപകരം, അതേ ഫലമുള്ള സംസ്ഥാന സ്റ്റോറുകളിൽ നിങ്ങൾക്ക് അന്വേഷിക്കാവുന്നതാണ്. സംസ്ഥാന സ്റ്റോറുകളിലേക്കുള്ള നേരിട്ടുള്ള അന്വേഷണങ്ങൾ അർത്ഥമാക്കുന്നത് കുറഞ്ഞ കോഡും (ഉപഭോക്താവില്ല) കുറഞ്ഞ സോഫ്റ്റ്‌വെയറും (ഫലങ്ങൾ സംഭരിക്കുന്നതിന് ഒരു ഡാറ്റാബേസ് പട്ടികയുടെ ആവശ്യമില്ല).

ഈ അധ്യായത്തിൽ ഞങ്ങൾ കുറച്ച് കാര്യങ്ങൾ ഉൾപ്പെടുത്തിയിട്ടുണ്ട്, അതിനാൽ സംസ്ഥാന സ്റ്റോറുകൾക്കെതിരായ സംവേദനാത്മക ചോദ്യങ്ങളെക്കുറിച്ചുള്ള ഞങ്ങളുടെ ചർച്ച ഞങ്ങൾ ഇപ്പോൾ ഉപേക്ഷിക്കും. എന്നാൽ വിഷമിക്കേണ്ട: 9-ാം അധ്യായത്തിൽ, സംവേദനാത്മക ചോദ്യങ്ങളുള്ള ഒരു ലളിതമായ ഡാഷ്‌ബോർഡ് ആപ്ലിക്കേഷൻ ഞങ്ങൾ സൃഷ്ടിക്കും. സംവേദനാത്മക അന്വേഷണങ്ങളും നിങ്ങൾക്ക് അവ എങ്ങനെ കാഫ്ക സ്ട്രീം ആപ്ലിക്കേഷനുകളിലേക്ക് ചേർക്കാമെന്നും കാണിക്കുന്നതിന് ഇതിൽ നിന്നും മുമ്പത്തെ അധ്യായങ്ങളിൽ നിന്നുമുള്ള ചില ഉദാഹരണങ്ങൾ ഇത് ഉപയോഗിക്കും.

സംഗ്രഹം

• KStream ഒബ്‌ജക്റ്റുകൾ ഒരു ഡാറ്റാബേസിലേക്കുള്ള ഇൻസെർട്ടുകളുമായി താരതമ്യപ്പെടുത്താവുന്ന ഇവന്റുകളുടെ സ്ട്രീമുകളെ പ്രതിനിധീകരിക്കുന്നു. KTable ഒബ്‌ജക്റ്റുകൾ ഒരു ഡാറ്റാബേസിലേക്കുള്ള അപ്‌ഡേറ്റുകൾ പോലെയുള്ള അപ്‌ഡേറ്റ് സ്ട്രീമുകളെ പ്രതിനിധീകരിക്കുന്നു. KTable ഒബ്‌ജക്റ്റിന്റെ വലുപ്പം വളരുന്നില്ല, പഴയ റെക്കോർഡുകൾ പുതിയവ ഉപയോഗിച്ച് മാറ്റിസ്ഥാപിക്കുന്നു.
• അഗ്രഗേഷൻ പ്രവർത്തനങ്ങൾക്ക് KTable വസ്തുക്കൾ ആവശ്യമാണ്.
• വിൻഡോയിംഗ് പ്രവർത്തനങ്ങൾ ഉപയോഗിച്ച്, നിങ്ങൾക്ക് സമാഹരിച്ച ഡാറ്റയെ ടൈം ബക്കറ്റുകളായി വിഭജിക്കാം.
• GlobalKTable ഒബ്‌ജക്‌റ്റുകൾക്ക് നന്ദി, പാർട്ടീഷനിംഗ് പരിഗണിക്കാതെ തന്നെ നിങ്ങൾക്ക് ആപ്ലിക്കേഷനിൽ എവിടെയും റഫറൻസ് ഡാറ്റ ആക്‌സസ് ചെയ്യാൻ കഴിയും.
• KStream, KTable, GlobalKTable ഒബ്‌ജക്റ്റുകൾ എന്നിവ തമ്മിലുള്ള കണക്ഷനുകൾ സാധ്യമാണ്.

ഇതുവരെ, ഉയർന്ന തലത്തിലുള്ള KStream DSL ഉപയോഗിച്ച് കാഫ്ക സ്ട്രീം ആപ്ലിക്കേഷനുകൾ നിർമ്മിക്കുന്നതിലാണ് ഞങ്ങൾ ശ്രദ്ധ കേന്ദ്രീകരിച്ചത്. വൃത്തിയും സംക്ഷിപ്തവുമായ പ്രോഗ്രാമുകൾ സൃഷ്ടിക്കാൻ ഉയർന്ന തലത്തിലുള്ള സമീപനം നിങ്ങളെ അനുവദിക്കുന്നുണ്ടെങ്കിലും, അത് ഉപയോഗിക്കുന്നത് ഒരു ട്രേഡ് ഓഫിനെ പ്രതിനിധീകരിക്കുന്നു. DSL KStream-ൽ പ്രവർത്തിക്കുക എന്നതിനർത്ഥം നിയന്ത്രണത്തിന്റെ അളവ് കുറയ്ക്കുന്നതിലൂടെ നിങ്ങളുടെ കോഡിന്റെ സംക്ഷിപ്തത വർദ്ധിപ്പിക്കുക എന്നാണ്. അടുത്ത അധ്യായത്തിൽ, ഞങ്ങൾ ലോ-ലെവൽ ഹാൻഡ്‌ലർ നോഡ് API നോക്കുകയും മറ്റ് ട്രേഡ് ഓഫുകൾ പരീക്ഷിക്കുകയും ചെയ്യും. പ്രോഗ്രാമുകൾ മുമ്പത്തേതിനേക്കാൾ ദൈർഘ്യമേറിയതായിരിക്കും, എന്നാൽ ഞങ്ങൾക്ക് ആവശ്യമായ ഏത് ഹാൻഡ്‌ലർ നോഡും സൃഷ്‌ടിക്കാൻ കഴിയും.

→ പുസ്തകത്തെക്കുറിച്ചുള്ള കൂടുതൽ വിശദാംശങ്ങൾ ഇവിടെ കാണാം പ്രസാധകന്റെ വെബ്സൈറ്റ്

→ ഹബ്രോജിറ്റെലിക്ക് കൂപ്പൺ ഉപയോഗിച്ച് 25% കിഴിവ് - കാഫ്ക സ്ട്രീമുകൾ

→ പുസ്തകത്തിന്റെ പേപ്പർ പതിപ്പിന് പണം നൽകുമ്പോൾ, ഒരു ഇലക്ട്രോണിക് പുസ്തകം ഇ-മെയിൽ വഴി അയയ്ക്കും.

അവലംബം: www.habr.com