विभिन्न स्रोतों से डेटा का बहुप्रचारित और सामंजस्य

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

वितरित प्रणालियों की विविधता को देखते हुए, लक्ष्य भंडारण में सत्यापित जानकारी की उपलब्धता डेटा स्थिरता के लिए एक महत्वपूर्ण मानदंड है।

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

इस मामले को अच्छे पुराने पायथन पर कैसे लागू किया जाए - इसे कटौती के तहत पढ़ें! चलो चलते हैं!


(छवि स्रोत)

परिचय

आइए कल्पना करें कि एक वित्तीय संस्थान में कई वितरित प्रणालियां हैं और हमें इन प्रणालियों में लेनदेन की पुष्टि करने और मेल खाते डेटा को लक्ष्य भंडारण में अपलोड करने के कार्य का सामना करना पड़ रहा है।

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

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

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

विकास के तहत प्रणाली का आर्किटेक्चर

उपयोग किए गए घटक:

• यादृच्छिक डेटा जनरेटर - एक पायथन स्क्रिप्ट जो एक सीएसवी फ़ाइल उत्पन्न करती है और इसके आधार पर एक डेटाबेस में एक तालिका भरती है;
• डेटा स्रोत - PostgreSQL डेटाबेस में CSV फ़ाइल और तालिका;
• एडेप्टर - इस मामले में, हम दो एडेप्टर का उपयोग करते हैं जो अपने स्रोतों (सीएसवी या डीबी) से डेटा निकालेंगे और मध्यवर्ती डेटाबेस में जानकारी दर्ज करेंगे;
• डेटाबेस - तीन टुकड़ों की मात्रा में: कच्चा डेटा, एक मध्यवर्ती डेटाबेस जो एडेप्टर द्वारा कैप्चर की गई जानकारी को संग्रहीत करता है, और दोनों स्रोतों से एक "क्लीन" डेटाबेस जिसमें सामंजस्यपूर्ण लेनदेन होता है।

प्रारंभिक प्रशिक्षण

डेटा स्टोरेज टूल के रूप में, हम डॉकटर कंटेनर में PostgreSQL डेटाबेस का उपयोग करेंगे और कंटेनर में चल रहे pgAdmin के माध्यम से अपने डेटाबेस के साथ बातचीत करेंगे:

docker run --name pg -d -e "POSTGRES_USER=my_user" -e "POSTGRES_PASSWORD=my_password" postgres 

चल रहा है pgAdmin:

 docker run -p 80:80 -e "PGADMIN_DEFAULT_EMAIL=user@domain.com" -e "PGADMIN_DEFAULT_PASSWORD=12345" -d dpage/pgadmin4 

सब कुछ शुरू हो जाने के बाद, कॉन्फ़िगरेशन फ़ाइल (conf / db.ini) को डेटाबेस में कनेक्शन स्ट्रिंग (एक प्रशिक्षण उदाहरण के लिए, आप कर सकते हैं!) निर्दिष्ट करना न भूलें।

 [POSTGRESQL] db_url=postgresql://my_user:my_password@172.17.0.2:5432/my_user 

सिद्धांत रूप में, एक कंटेनर का उपयोग वैकल्पिक है और आप अपने डेटाबेस सर्वर का उपयोग कर सकते हैं।

इनपुट जनरेशन

पायथन लिपि Gener_test_data परीक्षण डेटा की पीढ़ी के लिए जिम्मेदार है, जो उत्पन्न करने के लिए वांछित संख्या में प्रविष्टियां लेती है। GenerateTestData वर्ग के मुख्य कार्य द्वारा संचालन के अनुक्रम का आसानी से पता लगाया जा सकता है:

  @m.timing def run(self, num_rows): """ Run the process """ m.info('START!') self.create_db_schema() self.create_folder('data') self.create_csv_file(num_rows) self.bulk_copy_to_db() self.random_delete_rows() self.random_update_rows() m.info('END!') 

इसलिए, फ़ंक्शन निम्न चरणों का पालन करता है:

• डेटाबेस में स्कीमा बनाना (हम सभी मुख्य स्कीमा और टेबल बनाते हैं);
• एक परीक्षण फ़ाइल संग्रहीत करने के लिए एक फ़ोल्डर बनाना;
• दी गई संख्या के साथ एक परीक्षण फ़ाइल बनाना;
• लक्ष्य तालिका लेन-देन_db_raw.transaction_log में बल्क-सम्मिलित डेटा;
• इस तालिका में कई पंक्तियों का आकस्मिक विलोपन;
• इस तालिका में कई पंक्तियों का यादृच्छिक अद्यतन।

विलोपन और संशोधन आवश्यक है ताकि तुलना की गई वस्तुओं में कम से कम कुछ विसंगति हो। इन विसंगतियों की तलाश करने में सक्षम होना महत्वपूर्ण है!

 @m.timing @m.wrapper(m.entering, m.exiting) def random_delete_rows(self): """ Random deleting some rows from the table """ sql_command = sql.SQL(""" delete from {0}.{1} where ctid = any(array( select ctid from {0}.{1} tablesample bernoulli (1) ))""").format(sql.Identifier(self.schema_raw), sql.Identifier(self.raw_table_name)) try: rows = self.database.execute(sql_command) m.info('Has been deleted [%s rows] from table %s' % (rows, self.raw_table_name)) except psycopg2.Error as err: m.error('Oops! Delete random rows has been FAILED. Reason: %s' % err.pgerror) @m.timing @m.wrapper(m.entering, m.exiting) def random_update_rows(self): """ Random update some rows from the table """ sql_command = sql.SQL(""" update {0}.{1} set transaction_amount = round(random()::numeric, 2) where ctid = any(array( select ctid from {0}.{1} tablesample bernoulli (1) ))""").format(sql.Identifier(self.schema_raw), sql.Identifier(self.raw_table_name)) try: rows = self.database.execute(sql_command) m.info('Has been updated [%s rows] from table %s' % (rows, self.raw_table_name)) except psycopg2.Error as err: m.error('Oops! Delete random rows has been FAILED. Reason: %s' % err.pgerror) 

सीएसवी प्रारूप में एक पाठ फ़ाइल में एक परीक्षण डेटा सेट और बाद की रिकॉर्डिंग की पीढ़ी इस प्रकार है:

• एक यादृच्छिक लेनदेन यूआईडी बनाया गया है;
• एक यादृच्छिक यूआईडी खाता संख्या बनाई जाती है (डिफ़ॉल्ट रूप से, हम दस अद्वितीय खाते लेते हैं, लेकिन "random_accounts" पैरामीटर को बदलकर कॉन्फ़िगरेशन फ़ाइल का उपयोग करके इस मान को बदला जा सकता है);
• लेन-देन की तारीख - कॉन्फ़िगरेशन फ़ाइल (initial_date) में निर्दिष्ट तिथि से एक यादृच्छिक तिथि;
• लेनदेन का प्रकार (लेनदेन / कमीशन);
• लेन-देन की राशि;
• डेटा जनरेशन में मुख्य कार्य TestDataCadoror की जनरेट_टैस्ट_डाटा_बाई_चंक विधि द्वारा किया जाता है:

 @m.timing def generate_test_data_by_chunk(self, chunk_start, chunk_end): """ Generating and saving to the file """ num_rows_mp = chunk_end - chunk_start new_rows = [] for _ in range(num_rows_mp): transaction_uid = uuid.uuid4() account_uid = choice(self.list_acc) transaction_date = (self.get_random_date(self.date_in, 0) .__next__() .strftime('%Y-%m-%d %H:%M:%S')) type_deal = choice(self.list_type_deal) transaction_amount = randint(-1000, 1000) new_rows.append([transaction_uid, account_uid, transaction_date, type_deal, transaction_amount]) self.write_in_file(new_rows, chunk_start, chunk_end) 

इस फ़ंक्शन की एक विशेषता कई समानांतर असिंक्रोनस प्रक्रियाओं में लॉन्च है, जिनमें से प्रत्येक 50K रिकॉर्ड के अपने हिस्से को उत्पन्न करता है। यह "चिप" आपको कई मिलियन लाइनों पर एक फ़ाइल बनाने की अनुमति देगा जल्दी से पर्याप्त

 def run_csv_writing(self): """ Writing the test data into csv file """ pool = mp.Pool(mp.cpu_count()) jobs = [] for chunk_start, chunk_end in self.divide_into_chunks(0, self.num_rows): jobs.append(pool.apply_async(self.generate_test_data_by_chunk, (chunk_start, chunk_end))) # wait for all jobs to finish for job in jobs: job.get() # clean up pool.close() pool.join() 

पाठ फ़ाइल के पूरा होने के बाद, बल्क_इन्टर कमांड को संसाधित किया जाता है और इस फ़ाइल से सभी डेटा लेनदेन_db_raw.transaction_log तालिका में आते हैं ।

इसके अलावा, दो स्रोतों में बिल्कुल समान डेटा होगा और सामंजस्य कुछ भी दिलचस्प नहीं लगेगा, इसलिए हम डेटाबेस में कई यादृच्छिक पंक्तियों को हटाते हैं और संशोधित करते हैं।

स्क्रिप्ट चलाएँ और 10K लाइनों पर लेनदेन के साथ एक परीक्षण CSV फ़ाइल बनाएँ:

 ./generate_test_data.py 10000 

स्क्रीनशॉट से पता चलता है कि 10K लाइनों की एक फ़ाइल प्राप्त हुई थी, 10K को डेटाबेस में लोड किया गया था, लेकिन फिर डेटाबेस से 112 लाइनें हटा दी गईं और एक अन्य 108 को बदल दिया गया। परिणाम: फ़ाइल और डेटाबेस में तालिका 220 प्रविष्टियों द्वारा भिन्न होती है।

"ठीक है, बहु कहां है?", आप पूछते हैं।
और इसका काम तब देखा जा सकता है जब आप 10K रिकॉर्ड से नहीं, बल्कि उदाहरण के लिए, 1M से बड़ी फ़ाइल बनाते हैं। क्या हम कोशिश करेंगे?

 ./generate_test_data.py 1000000 

डेटा को लोड करने, हटाने और यादृच्छिक रिकॉर्ड बदलने के बाद, हम टेबल से पाठ फ़ाइल के अंतर को देखते हैं: 19,939 पंक्तियाँ (जिनमें से 10,022 को यादृच्छिक रूप से हटा दिया गया, और 9,917 बदल गए)।

चित्र दिखाता है कि अभिलेखों की पीढ़ी अतुल्यकालिक, असंगत थी। इसका मतलब यह है कि अगली प्रक्रिया स्टार्ट ऑर्डर को ध्यान में रखे बिना शुरू हो सकती है जैसे ही पिछले एक पूरा हो जाता है। इस बात की कोई गारंटी नहीं है कि परिणाम इनपुट के समान क्रम में होगा।

सीएसवी के लिए एडाप्टर

यह एडेप्टर केवल पहले कॉलम को छोड़कर, लेन-देन पहचानकर्ता को अपरिवर्तित करता है, अपरिवर्तित होता है और प्राप्त डेटा को डेटा / transaction_hashed.csv फ़ाइल में सहेजता है। उनके काम का अंतिम चरण इस फाइल को COPY कमांड का उपयोग करके सामंजस्य_ db स्कीमा की अस्थायी तालिका में लोड करना है ।

इष्टतम फ़ाइल रीडिंग कई समानांतर प्रक्रियाओं द्वारा किया जाता है। हम 5 मेगाबाइट्स के टुकड़ों में, लाइन द्वारा लाइन पढ़ते हैं। आंकड़ा "5 मेगाबाइट" अनुभवजन्य विधि द्वारा प्राप्त किया गया था। यह टेक्स्ट के एक टुकड़े के इस आकार के साथ था कि हम अपनी आभासी मशीन पर बड़ी फ़ाइलों को पढ़ने के लिए सबसे छोटा समय प्राप्त करने में सक्षम थे। आप इस पैरामीटर के साथ अपने पर्यावरण पर प्रयोग कर सकते हैं और देख सकते हैं कि ऑपरेटिंग समय कैसे बदलेगा:

 @m.timing def process_wrapper(self, chunk_start, chunk_size): """ Read a particular chunk """ with open(self.file_name_raw, newline='\n') as file: file.seek(chunk_start) lines = file.read(chunk_size).splitlines() for line in lines: self.process(line) def chunkify(self, size=1024*1024*5): """ Return a new chunk """ with open(self.file_name_raw, 'rb') as file: chunk_end = file.tell() while True: chunk_start = chunk_end file.seek(size, 1) file.readline() chunk_end = file.tell() if chunk_end > self.file_end: chunk_end = self.file_end yield chunk_start, chunk_end - chunk_start break else: yield chunk_start, chunk_end - chunk_start @m.timing def run_reading(self): """ The main method for the reading """ # init objects pool = mp.Pool(mp.cpu_count()) jobs = [] m.info('Run csv reading...') # create jobs for chunk_start, chunk_size in self.chunkify(): jobs.append(pool.apply_async(self.process_wrapper, (chunk_start, chunk_size))) # wait for all jobs to finish for job in jobs: job.get() # clean up pool.close() pool.join() m.info('CSV file reading has been completed') 

1M रिकॉर्ड पर पहले से बनाई गई फ़ाइल पढ़ने का उदाहरण:


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

प्रत्येक सुलह प्रक्रिया के लिए एक अद्वितीय नाम के साथ एक अस्थायी तालिका का उपयोग करना आपको अतिरिक्त रूप से एक डेटाबेस में सुलह प्रक्रिया को समानांतर करने की अनुमति देता है।

PostgreSQL के लिए एडाप्टर

तालिका में संग्रहीत डेटा को संसाधित करने के लिए एडाप्टर फ़ाइल के लिए एडाप्टर के रूप में लगभग एक ही तर्क काम करता है:

• तालिका के कुछ हिस्सों में पढ़ना (यदि यह बड़ी है, 100K प्रविष्टियों से अधिक) और लेनदेन पहचानकर्ता को छोड़कर सभी कॉलमों के लिए हैश लेना;
• तब सामंजस्य_db तालिका में संसाधित डेटा का सम्मिलन होता है । भंडारण _ $ (int (time.time ()) ।

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

तालिका के आकार के आधार पर, प्रसंस्करण के लिए आवश्यक प्रक्रियाओं की संख्या की गणना की जाती है और प्रत्येक प्रक्रिया के भीतर 10 कार्यों में एक विभाजन होता है।

 def read_data(self): """ Read the data from the postgres and shared those records with each processor to perform their operation using threads """ threads_array = self.get_threads(0, self.max_id_num_row, self.pid_max) for pid in range(1, len(threads_array) + 1): m.info('Process %s' % pid) # Getting connection from the connection pool select_conn = self._select_conn_pool.getconn() select_conn.autocommit = 1 # Creating 10 process to perform the operation process = Process(target=self.process_data, args=(self.data_queque, pid, threads_array[pid-1][0], threads_array[pid-1][1], select_conn)) process.daemon = True process.start() process.join() select_conn.close() 

विसंगतियों के लिए खोजें

हम दो एडेप्टर से प्राप्त डेटा के सत्यापन के लिए आगे बढ़ते हैं।

सुलह (या एक विसंगति रिपोर्ट प्राप्त) डेटाबेस की ओर से SQL भाषा की सभी शक्ति का उपयोग करके होती है।

SQL क्वेरी काफी सीधी है - यह सिर्फ एक तालिका है जिसमें एडाप्टर्स से ट्रांजेक्शन आईडी के साथ डेटा शामिल होता है:

 sql_command = sql.SQL(""" select s1.adapter_name, count(s1.transaction_uid) as tran_count from {0}.{1} s1 full join {0}.{1} s2 on s2.transaction_uid = s1.transaction_uid and s2.adapter_name != s1.adapter_name and s2.hash = s1.hash where s2.transaction_uid is null group by s1.adapter_name;""").format(sql.Identifier(self.schema_target), sql.Identifier(self.storage_table)) 

आउटपुट एक रिपोर्ट है:


जांचें कि क्या ऊपर की तस्वीर में सब कुछ सही है। हमें याद है कि डेटाबेस में तालिका से 9917 हटा दिए गए थे और 10,022 पंक्तियों को बदल दिया गया था। कुल 19939 लाइनें, जो रिपोर्ट में स्पष्ट है।

सारांश तालिका

यह स्टोरेज टेबल में "क्लीन" ट्रांजैक्शन डालने के लिए ही रहता है, जो अलग-अलग एडेप्टर में सभी तरह से मेल खाता है। यह प्रक्रिया निम्न SQL क्वेरी द्वारा की जाती है:

 sql_command = sql.SQL(""" with reconcil_data as ( select s1.transaction_uid from {0}.{1} s1 join {0}.{1} s2 on s2.transaction_uid = s1.transaction_uid and s2.adapter_name != s1.adapter_name where s2.hash = s1.hash and s1.adapter_name = 'postresql_adapter' ) insert into {2}.transaction_log select t.transaction_uid, t.account_uid, t.transaction_date, t.type_deal, t.transaction_amount from {3}.transaction_log t join reconcil_data r on t.transaction_uid = r.transaction_uid where not exists ( select 1 from {2}.transaction_log tl where tl.transaction_uid = t.transaction_uid ) """).format(sql.Identifier(self.schema_target), sql.Identifier(self.storage_table), sql.Identifier(self.schema_db_clean), sql.Identifier(self.schema_raw)) 

अस्थायी तालिका जिसे हमने एडेप्टर से डेटा के मध्यवर्ती भंडारण के रूप में उपयोग किया था, को हटाया जा सकता है।

निष्कर्ष

किए गए कार्य के दौरान, विभिन्न स्रोतों से डेटा को समेटने की एक प्रणाली विकसित की गई थी: एक पाठ फ़ाइल और डेटाबेस में एक तालिका। कम से कम अतिरिक्त उपकरणों का इस्तेमाल किया।

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

पूरे प्रोजेक्ट का सोर्स कोड GitHub पर मेरी रिपॉजिटरी में निहित है , मेरा सुझाव है कि आप इसके साथ खुद को परिचित करें।

मुझे सभी सवालों के जवाब देने और आपकी टिप्पणियों से परिचित होने में खुशी होगी।

मैं आपको सफलता की कामना करता हूं!