👈🏼 🛸 🎵 हम कई विश्लेषकों को अपने हाथ मुक्त करते हैं: ठेठ बैंकिंग कार्यों के स्वचालन के लिए एपीआई लिवी ☦️ 👟 🎮

नमस्कार, हेब्र!

यह कोई रहस्य नहीं है कि बैंक ग्राहकों की सॉल्वेंसी का आकलन करने के लिए विभिन्न स्रोतों (क्रेडिट ब्यूरो, मोबाइल ऑपरेटरों, आदि) से डेटा का उपयोग करते हैं। बाहरी भागीदारों की संख्या कई दर्जन तक पहुंच सकती है, और हमारी टीम में विश्लेषक केवल कुछ लोगों की भर्ती करेंगे। समस्या एक छोटी टीम के काम का अनुकूलन करने और नियमित कार्यों को कंप्यूटिंग सिस्टम में स्थानांतरित करने से उत्पन्न होती है।

हम विश्लेषण करेंगे कि यह डेटा बैंक में कैसे जाता है, और विश्लेषकों की एक टीम इस प्रक्रिया की निगरानी कैसे करती है।

चलिए शुरू करते हैं।

Hadoop पर आधारित हमारी वितरित प्रणाली और इससे जुड़ी सभी प्रक्रियाएं, हम संक्षेप में SmartData कहते हैं। स्मार्टडाटा बाहरी एजेंटों से एपीआई डेटा प्राप्त करता है। (इसके अलावा, इसके लिए एजेंट बैंक के बाहरी भागीदार और आंतरिक सिस्टम दोनों हैं)। बेशक, प्रत्येक ग्राहक के लिए कुछ "प्रासंगिक प्रोफ़ाइल" एकत्र करना उपयोगी होगा, जो हम करते हैं। स्रोतों से अपडेट किया गया डेटा ऑपरिटिफ़िल में आता है। आपरेटिफ़ाइल ग्राहक 360 के विचार को लागू करता है और इसे Hbase तालिकाओं के रूप में संग्रहीत किया जाता है। यह क्लाइंट के साथ आगे के काम के लिए सुविधाजनक है।

ग्राहक 360

ग्राहक 360 - संगठन में सभी प्रक्रियाओं में उपयोग किए जाने वाले क्लाइंट डेटा की सभी प्रकार की विशेषताओं के साथ परिचालन भंडारण को लागू करने के लिए एक दृष्टिकोण जो क्लाइंट और उसके डेटा के साथ काम करता है, जो क्लाइंट की कुंजी द्वारा सुलभ है।

एजेंटों के साथ काम जारी है और इसे नियंत्रित करने की आवश्यकता है। बातचीत और हिट दर की गुणवत्ता की जांच करने के लिए, साथ ही इस जानकारी को अन्य टीमों में स्थानांतरित करने में आसानी के लिए, हम विज़ुअलाइज़ेशन का उपयोग करते हैं, उदाहरण के लिए, झांकी में रिपोर्ट।

स्रोत डेटा को काफ्का को भेजा जाता है, जो पूर्व-संसाधित है और एचडीएफएस के आधार पर बनाए गए डेटालेक में रखा गया है। यह एक समाधान के साथ आने के लिए आवश्यक था कि एचडीएफएस से लॉग फाइलों की पार्सिंग को कैसे व्यवस्थित किया जाए, उनके प्रसंस्करण और विश्लेषणात्मक और विज़ुअलाइज़ेशन सिस्टम में दैनिक अपलोडिंग। और इसे पायथन लैपटॉप के लिए विश्लेषकों के प्यार के साथ भी जोड़ते हैं।

आंतरिक रसोई के साथ समाप्त करें और अभ्यास करने के लिए आगे बढ़ें।

हमारा समाधान लिवी एपीआई का उपयोग करना था। Livy आपको सीधे Jupyter से क्लस्टर में कोड सबमिट करने की अनुमति देता है। एक HTTP अनुरोध जिसमें पायथन (या स्काला) और मेटा डेटा में लिखा कोड लिवी को भेजा गया है। लिवी क्लस्टर पर स्पार्क सत्र के शुभारंभ की शुरुआत करता है, जिसे यार्न संसाधन प्रबंधक द्वारा प्रबंधित किया जाता है। HTTP अनुरोध भेजने के लिए अनुरोध मॉड्यूल उपयुक्त है। जो लोग साइटों को पार्स करना पसंद करते हैं, वे शायद पहले से ही उसे जानते हैं (और यदि नहीं, तो यहां उसके बारे में थोड़ा जानने का मौका है)।

हम आवश्यक मॉड्यूल आयात करते हैं और एक सत्र बनाते हैं। (हम अपने सत्र का पता भी तुरंत लगा लेते हैं, भविष्य में यह काम आएगा)। पैरामीटर में हम उपयोगकर्ता प्राधिकरण और स्क्रिप्ट भाषा के नाम के लिए डेटा पास करते हैं जिसे क्लस्टर निष्पादित करेगा।

import json, requests, schedule, time host = 'http://***:8998' data = {'kind': 'spark', 'proxyUser': 'user'} headers = {'Content-Type': 'application/json'} r = requests.post(host + '/sessions', data=json.dumps(data), headers=headers) session_id = r.json().get('id') print("session_id: " + str(session_id)) session_url = host + r.headers['location'] r = requests.get(session_url, headers=headers)

हम निष्क्रिय होने के लिए सत्र की स्थिति का इंतजार कर रहे हैं। यदि टाइमआउट सेट टाइमआउट से अधिक है - एक त्रुटि संदेश भेजें।

 timeout = time.time() + wait_time sess_state = ['starting', 'success', 'idle'] while(True): time.sleep(7) req_st = requests.get(session_url, headers=headers).json().get('state') if req_st != 'idle' and time.time() > timeout: requests.delete(session_url, headers=headers) send_message("Scheduler_error", req_st) break if req_st == 'idle': break if req_st not in sess_state: send_message("Scheduler_error", req_st) break print("Session_state: ", req_st)

अब आप Livy को कोड भेज सकते हैं।

 statements_url = session_url + '/statements' data = {'code': '1 + 1'} r = requests.post(statements_url, data=json.dumps(data), headers=headers) statement_url = host + r.headers['location'] r = requests.get(statement_url, headers=headers) while (requests.get(statement_url, headers=headers).json()['progress'] != 1): time.sleep(15) r = requests.get(statement_url, headers=headers).json()['output'] session_url = 'http://***:8998/sessions/' + str(session_id)

लूप में, हम कोड निष्पादन के अंत की प्रतीक्षा करते हैं, हमें प्रोसेसिंग परिणाम मिलता है:

 r.get('data').get('text/plain')

डिलीट मेथड सेशन हट जाएगा।

 requests.delete(session_url, headers=headers)

दैनिक उतराई के लिए, आप कई विकल्पों का उपयोग कर सकते हैं, उन्होंने पहले से ही हब पर क्रोन के बारे में लिखा था, लेकिन उपयोगकर्ता के अनुकूल शेड्यूल मॉड्यूल के बारे में - नहीं। बस इसे कोड में जोड़ें, इसे स्पष्टीकरण की आवश्यकता नहीं होगी। और, सुविधा के लिए, मैं सभी गणनाओं को एक स्थान पर एकत्र करूंगा।

कोड

 import json, requests, schedule, time schedule.every().day.at("16:05").do(job, 300) while True: schedule.run_pending() def job(wait_time): host = 'http://***:8998' data = {'kind': 'spark', 'proxyUser': 'user'} headers = {'Content-Type': 'application/json'} r = requests.post(host + '/sessions', data=json.dumps(data), headers=headers) session_id = r.json().get('id') print("session_id: " + str(session_id)) session_url = host + r.headers['location'] r = requests.get(session_url, headers=headers) timeout = time.time() + wait_time sess_state = ['starting', 'success', 'idle'] while(True): time.sleep(7) req_st = requests.get(session_url, headers=headers).json().get('state') if req_st != 'idle' and time.time() > timeout: requests.delete(session_url, headers=headers) break if req_st == 'idle': break if req_st not in sess_state: send_message("Scheduler_error", req_st) break print("Session_state: ", req_st) statements_url = session_url + '/statements' data = {'code': '1 + 1'} r = requests.post(statements_url, data=json.dumps(data),headers=headers) statement_url = host + r.headers['location'] r = requests.get(statement_url, headers=headers) while (requests.get(statement_url, headers=headers).json()['progress'] != 1): time.sleep(15) r = requests.get(statement_url, headers=headers).json()['output'] session_url = 'http://***:8998/sessions/' + str(session_id) print(r.get('data').get('text/plain')) #requests.delete(session_url, headers=headers) def send_message(subject, text): import smtplib from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText me = "my_email_adress" you = "email_adress" msg = MIMEMultipart('alternative') msg['Subject'] = subject msg['From'] = me msg['To'] = you text = text part1 = MIMEText(text, 'plain') msg.attach(part1) s = smtplib.SMTP('domain.org') s.ehlo() s.starttls() s.login("user", "password") s.sendmail(me, you, msg.as_string()) s.quit()

निष्कर्ष:

शायद यह समाधान सबसे अच्छा होने का दावा नहीं करता है, लेकिन यह विश्लेषकों की टीम के लिए पारदर्शी है। पेशेवरों है कि मैं इसे देख रहा हूँ:

स्वचालन के लिए परिचित ज्यूपिटर का उपयोग करने की क्षमता
दृश्य बातचीत
टीम के सदस्य को यह चुनने का अधिकार है कि वह फाइलों (स्पार्क-चिड़ियाघर) के साथ कैसे काम करेगा, नतीजतन, मौजूदा स्क्रिप्ट को फिर से लिखने की आवश्यकता नहीं है।

बेशक, बड़ी संख्या में कार्यों को शुरू करते समय, आपको मुक्त संसाधनों की निगरानी करनी होगी, अनलोडिंग के बीच संचार को कॉन्फ़िगर करना होगा। इन मुद्दों को व्यक्तिगत आधार पर हल किया जाता है और सहयोगियों के साथ सहमति व्यक्त की जाती है।

यह बहुत अच्छा होगा यदि कम से कम एक टीम इस निर्णय को नोट करती है।

संदर्भ

Livy प्रलेखन

हम कई विश्लेषकों को अपने हाथ मुक्त करते हैं: ठेठ बैंकिंग कार्यों के स्वचालन के लिए एपीआई लिवी

निष्कर्ष:

संदर्भ

More articles: