అపాచీ స్పార్క్ స్ట్రీమింగ్‌లో సంచిత స్టేట్‌ఫుల్ ట్రాన్స్ఫర్మేషన్

పృథ్వీరాజ్ బోస్ సహకారం అందించారు

నా మునుపటి బ్లాగులో నేను అపాచీ స్పార్క్ స్ట్రీమింగ్ యొక్క విండోస్ భావనను ఉపయోగించి రాష్ట్ర పరివర్తన గురించి చర్చించాను. మీరు దీన్ని చదువుకోవచ్చు ఇక్కడ .

పవర్ బైలో డాష్‌బోర్డ్‌ను ఎలా సృష్టించాలి

ఈ పోస్ట్‌లో నేను అపాచీ స్పార్క్ స్ట్రీమింగ్‌లో సంచిత స్థితిగతుల కార్యకలాపాలను చర్చించబోతున్నాను. మీరు స్పార్క్ స్ట్రీమింగ్‌కు కొత్తగా ఉంటే, విండోస్ ఎలా పనిచేస్తుందో అర్థం చేసుకోవడానికి నా మునుపటి బ్లాగును చదవమని నేను గట్టిగా సిఫార్సు చేస్తున్నాను.

స్పార్క్ స్ట్రీమింగ్‌లో స్టేట్‌ఫుల్ ట్రాన్స్ఫర్మేషన్ రకాలు (కొనసాగింపు…)

> సంచిత ట్రాకింగ్

మేము ఉపయోగించాము తగ్గించుబైకేఅండ్‌విండో (…) కీల స్థితులను ట్రాక్ చేయడానికి API, అయితే విండోస్ కొన్ని ఉపయోగ సందర్భాలకు పరిమితులను కలిగిస్తుంది. కీల విండోలను టైమ్ విండోకు పరిమితం చేయకుండా, అంతటా దాని యొక్క రాష్ట్రాలను కూడబెట్టుకోవాలనుకుంటే? అలాంటప్పుడు మనం ఉపయోగించాల్సి ఉంటుంది updateStateByKey (…) మంట.

ఈ API స్పార్క్ 1.3.0 లో ప్రవేశపెట్టబడింది మరియు ఇది చాలా ప్రజాదరణ పొందింది. అయితే ఈ API కి కొంత పనితీరు ఓవర్ హెడ్ ఉంది, కాలక్రమేణా రాష్ట్రాల పరిమాణం పెరిగే కొద్దీ దాని పనితీరు క్షీణిస్తుంది. ఈ API వినియోగాన్ని చూపించడానికి నేను ఒక నమూనా వ్రాసాను. మీరు కోడ్‌ను కనుగొనవచ్చు ఇక్కడ .

స్పార్క్ 1.6.0 కొత్త API ని ప్రవేశపెట్టింది mapWithState (…) ఇది ఎదురయ్యే పనితీరు ఓవర్‌హెడ్‌లను పరిష్కరిస్తుంది updateStateByKey (…) . ఈ బ్లాగులో నేను వ్రాసిన నమూనా ప్రోగ్రామ్‌ను ఉపయోగించి ఈ ప్రత్యేక API గురించి చర్చించబోతున్నాను. మీరు కోడ్‌ను కనుగొనవచ్చు ఇక్కడ .

నేను కోడ్ వాక్-త్రూలోకి ప్రవేశించే ముందు, చెక్‌పాయింటింగ్‌లో కొన్ని పదాలను వదిలివేద్దాం. ఏదైనా రాష్ట్ర పరివర్తన కోసం, చెక్‌పాయింటింగ్ తప్పనిసరి. చెక్ పాయింట్ అనేది డ్రైవర్ ప్రోగ్రామ్ విఫలమైతే కీల స్థితిని పునరుద్ధరించడానికి ఒక విధానం. డ్రైవర్ పున ar ప్రారంభించినప్పుడు, చెక్ పాయింట్ ఫైళ్ళ నుండి కీల స్థితి పునరుద్ధరించబడుతుంది. చెక్‌పాయింట్ స్థానాలు సాధారణంగా HDFS లేదా అమెజాన్ S3 లేదా ఏదైనా నమ్మదగిన నిల్వ. కోడ్‌ను పరీక్షిస్తున్నప్పుడు, స్థానిక ఫైల్ సిస్టమ్‌లో కూడా నిల్వ చేయవచ్చు.

నమూనా ప్రోగ్రామ్‌లో, మేము హోస్ట్ = లోకల్ హోస్ట్ మరియు పోర్ట్ = 9999 లో సాకెట్ టెక్స్ట్ స్ట్రీమ్‌ను వింటాము. ఇది ఇన్‌కమింగ్ స్ట్రీమ్‌ను టోకనైజ్ చేస్తుంది (పదాలు, సంఘటనల సంఖ్య) మరియు 1.6.0 API ని ఉపయోగించి పద గణనను ట్రాక్ చేస్తుంది mapWithState (…) . అదనంగా, నవీకరణలు లేని కీలను ఉపయోగించి తొలగించబడతాయి స్టేట్స్పెక్.టైమ్ అవుట్ API. మేము HDFS లో చెక్‌పాయింట్ చేస్తున్నాము మరియు ప్రతి 20 సెకన్లకు చెక్‌పాయింటింగ్ ఫ్రీక్వెన్సీ ఉంటుంది.

మొదట స్పార్క్ స్ట్రీమింగ్ సెషన్‌ను సృష్టించండి,

మేము ఒక సృష్టిస్తాము చెక్ పాయింట్ డిర్ HDFS లో ఆపై ఆబ్జెక్ట్ పద్ధతిని కాల్ చేయండి getOrCreate (…) . ది getOrCreate API తనిఖీ చేస్తుంది చెక్ పాయింట్ డిర్ పునరుద్ధరించడానికి మునుపటి రాష్ట్రాలు ఉన్నాయా అని చూడటానికి, అది ఉనికిలో ఉంటే, అది స్పార్క్ స్ట్రీమింగ్ సెషన్‌ను పున reat సృష్టిస్తుంది మరియు క్రొత్త డేటాతో వెళ్లడానికి ముందు ఫైల్‌లలో నిల్వ చేసిన డేటా నుండి కీల స్థితులను నవీకరిస్తుంది. లేకపోతే ఇది కొత్త స్పార్క్ స్ట్రీమింగ్ సెషన్‌ను సృష్టిస్తుంది.

ది getOrCreate చెక్‌పాయింట్ డైరెక్టరీ పేరు మరియు ఫంక్షన్‌ను తీసుకుంటుంది (వీటికి మేము పేరు పెట్టాము createFunc ) ఎవరి సంతకం ఉండాలి () => స్ట్రీమింగ్కాంటెక్స్ట్ .

లోపల కోడ్‌ను పరిశీలిద్దాం createFunc .

పంక్తి # 2: మేము ఉద్యోగ పేరుతో “TestMapWithStateJob” మరియు బ్యాచ్ విరామం = 5 సెకన్లకు స్ట్రీమింగ్ సందర్భాన్ని సృష్టిస్తాము.

పంక్తి # 5: చెక్‌పాయింట్ డైరెక్టరీని సెట్ చేయండి.

పంక్తి # 8: తరగతి ఉపయోగించి స్టేట్ స్పెసిఫికేషన్ సెట్ చేయండి org.apache.streaming.StateSpec వస్తువు. మేము మొదట రాష్ట్రాన్ని ట్రాక్ చేసే ఫంక్షన్‌ను సెట్ చేసాము, తరువాత వచ్చే పరివర్తనల సమయంలో ఉత్పత్తి చేయబడే DStreams కోసం విభజనల సంఖ్యను సెట్ చేస్తాము. చివరగా మేము సమయం ముగిసింది (30 సెకన్లకు), అక్కడ కీ కోసం ఏదైనా నవీకరణ 30 సెకన్లలో అందుకోకపోతే, కీ స్థితి తొలగించబడుతుంది.

12 వ పంక్తి: సాకెట్ స్ట్రీమ్‌ను సెటప్ చేయండి, ఇన్‌కమింగ్ బ్యాచ్ డేటాను చదును చేయండి, కీ-విలువ జతను సృష్టించండి, కాల్ చేయండి mapWithState , చెక్‌పాయింటింగ్ విరామాన్ని 20 లకు సెట్ చేసి, చివరకు ఫలితాలను ముద్రించండి.

స్పార్క్ ఫ్రేమ్‌వర్క్ వ e createFunc మునుపటి విలువ మరియు ప్రస్తుత స్థితితో ప్రతి కీ కోసం. మేము మొత్తాన్ని లెక్కించి, రాష్ట్రాన్ని సంచిత మొత్తంతో అప్‌డేట్ చేస్తాము మరియు చివరికి మేము కీ కోసం మొత్తాన్ని తిరిగి ఇస్తాము.

SQL లో తేదీ కోసం డేటా రకం

గితుబ్ సోర్సెస్ -> TestMapStateWithKey.scala , TestUpdateStateByKey.scala

మాకు ప్రశ్న ఉందా? దయచేసి దీన్ని వ్యాఖ్యల విభాగంలో పేర్కొనండి మరియు మేము మిమ్మల్ని సంప్రదిస్తాము.

సంబంధిత పోస్ట్లు:

అపాచీ స్పార్క్ & స్కేలాతో ప్రారంభించండి

స్పార్క్ స్ట్రీమింగ్‌లో విండోతో స్టేట్‌ఫుల్ ట్రాన్స్ఫర్మేషన్స్