🛀🏿 🤔 🌪️ Apache बीम SDK और Python का उपयोग करके GCP पर आधारित Pub / Sub से BigQuery तक डेटा स्ट्रीमिंग के लिए एक डेटाफ्लो टेम्प्लेट बनाना 🏌️ 🏺 💭

फिलहाल, मैं डेटा को स्ट्रीमिंग (और परिवर्तित) करने के कार्य में लगा हूं। कुछ हलकों में
ऐसी प्रक्रिया को ETL के रूप में जाना जाता है, अर्थात जानकारी का निष्कर्षण, रूपांतरण और लोडिंग।

संपूर्ण प्रक्रिया में निम्न Google क्लाउड प्लेटफ़ॉर्म सेवाओं की भागीदारी शामिल है:

पब / उप - रियलटाइम डेटा स्ट्रीमिंग के लिए सेवा
Dataflow - डेटा परिवर्तित करने के लिए एक सेवा (कर सकते हैं)
रियलटाइम और बैच मोड दोनों में काम करें)
BigQuery - तालिकाओं के रूप में डेटा संग्रहीत करने के लिए एक सेवा
(एसक्यूएल का समर्थन करता है)

0. वर्तमान स्थिति

फिलहाल, उपरोक्त सेवाओं पर स्ट्रीमिंग का एक कार्यशील संस्करण है, हालाँकि इसमें
एक टेम्पलेट के रूप में, मानक लोगों में से एक का उपयोग किया जाता है ।

समस्या यह है कि यह टेम्पलेट 1 से 1 डेटा ट्रांसफर प्रदान करता है, अर्थात पर
पब / सब के प्रवेश द्वार पर हमारे पास एक JSON प्रारूप स्ट्रिंग है, आउटपुट पर हमारे पास खेतों के साथ एक BigQuery तालिका है:
जो JSON के शीर्ष स्तर पर ऑब्जेक्ट्स की कुंजियों के अनुरूप है।

1. समस्या का विवरण

एक डेटाफ्लो टेम्प्लेट बनाएं जो आपको आउटपुट पर एक टेबल या टेबल प्राप्त करने की अनुमति देता है
दी गई शर्तों के अनुसार। उदाहरण के लिए, हम प्रत्येक के लिए एक अलग तालिका बनाना चाहते हैं
एक विशिष्ट इनपुट JSON कुंजी के मान। इस तथ्य को ध्यान में रखना आवश्यक है कि कुछ
इनपुट JSON ऑब्जेक्ट्स में NSON J मान के रूप में हो सकता है, अर्थात आवश्यक है
नेस्टेड भंडारण के लिए RECORD के प्रकार के साथ BigQuery टेबल बनाने में सक्षम हो
डेटा।

2. निर्णय की तैयारी

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

हालांकि, इस तथ्य के कारण कि मैं जावा से परिचित नहीं हूं, मैंने पायथन का उपयोग किया।

इससे पहले कि आप एक टेम्पलेट बनाना शुरू करें, आपके पास निम्नलिखित होना चाहिए:

इनपुट JSON प्रारूप और इसे समय में नहीं बदलना चाहिए
स्कीमा या BigQuery टेबल के स्कीमा जिसमें डेटा प्रवाहित किया जाएगा
तालिकाओं की संख्या जिसमें आउटपुट डेटा स्ट्रीम स्ट्रीम की जाएगी

ध्यान दें कि एक टेम्पलेट बनाने और उसके आधार पर Dataflow Job शुरू करने के बाद, ये पैरामीटर हो सकते हैं
नया टेम्प्लेट बनाकर ही बदलें।

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

3. निर्णय

प्रक्रिया की अधिक संपूर्ण समझ के लिए, यह तथाकथित पाइपलाइन का आरेख लाने के लायक है
अपाचे बीम प्रलेखन से।

हमारे मामले में (हम कई तालिकाओं में विभाजन का उपयोग करेंगे):

इनपुट - डेटा डाटाफ्लो जॉब में PubSub से आता है
ट्रांसफ़र # 1 - डेटा को स्ट्रिंग से पायथन डिक्शनरी में परिवर्तित किया जाता है, हमें आउटपुट मिलता है
PCollection # 1
ट्रांसफ़ॉर्म # 2 - डेटा को अलग-अलग तालिकाओं के अनुसार अलग करने के लिए, में टैग किया गया है
उत्पादन PCollection # 2 है (वास्तव में एक PCollection tuple)
ट्रांसफ़ॉर्म # 3 - PCollection # 2 से डेटा योजनाओं का उपयोग कर तालिकाओं को लिखा जाता है
तालिका बनाना

अपने स्वयं के टेम्पलेट को लिखने की प्रक्रिया में, मैं इन उदाहरणों से सक्रिय रूप से प्रेरित था।

टिप्पणियों के साथ टेम्पलेट कोड (पिछले लेखकों से उसी तरह टिप्पणी छोड़ दी गई है):

 # coding=utf-8 from __future__ import absolute_import import logging import json import os import apache_beam as beam from apache_beam.pvalue import TaggedOutput from apache_beam.options.pipeline_options import PipelineOptions from apache_beam.options.pipeline_options import SetupOptions from apache_beam.options.pipeline_options import StandardOptions from apache_beam.io.gcp.bigquery import parse_table_schema_from_json #  GCP  gcp_project = '' #  Pub/Sub  topic_name = '' # Pub/Sub    'projects/_GCP_/topics/_' input_topic = 'projects/%s/topics/%s' % (gcp_project, topic_name) #  BigQuery  bq_dataset = 'segment_eu_test' #       schema_dir = './' class TransformToBigQuery(beam.DoFn): #          ,   # BigQuery IO     python dict def process(self, element, *args, **kwargs): body = json.loads(element) #       ,      # python dict       ,     #   yield body class TagDataWithReqType(beam.DoFn): #      , ..      #     ,       #  with_outputs + default def process(self, element, *args, **kwargs): req_type = element.get('_') types = ( 'type1', 'type2', 'type3', ) if req_type in types: yield TaggedOutput(req_type, element) else: yield element def run(): #       _.json   schema_dir,  #         ()  schema_dct = {} for schema_file in os.listdir(schema_dir): filename_list = schema_file.split('.') if filename_list[-1] == 'json': with open('%s/%s' % (schema_dir, schema_file)) as f: schema_json = f.read() schema_dct[filename_list[0]] = json.dumps({'fields': json.loads(schema_json)}) # We use the save_main_session option because one or more DoFn's in this # workflow rely on global context (eg, a module imported at module level). pipeline_options = PipelineOptions() p = beam.Pipeline(options=pipeline_options) pipeline_options.view_as(SetupOptions).save_main_session = True pipeline_options.view_as(StandardOptions).streaming = True # Read from PubSub into a PCollection. input_stream = p | beam.io.ReadFromPubSub(input_topic) # Transform stream to BigQuery IO format stream_bq = input_stream | 'transform to BigQuery' >> beam.ParDo(TransformToBigQuery()) # Tag stream by schema name tagged_stream = \ stream_bq \ | 'tag data by type' >> beam.ParDo(TagDataWithReqType()). with_outputs(*schema_dct.keys(), main='default') # Stream unidentified data to default table tagged_stream.default | 'push to default table' >> beam.io.WriteToBigQuery( '%s:%s.default' % ( gcp_project, bq_dataset, ), schema=parse_table_schema_from_json(schema_dct.get('default')), ) # Stream data to BigQuery tables by number of schema names for name, schema in schema_dct.iteritems(): tagged_stream[name] | 'push to table %s' % name >> beam.io.WriteToBigQuery( '%s:%s.%s' % ( gcp_project, bq_dataset, name), schema=parse_table_schema_from_json(schema), ) result = p.run() result.wait_until_finish() if __name__ == '__main__': logging.getLogger().setLevel(logging.INFO) logger = logging.getLogger(__name__) run()

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

 pipeline_options = PipelineOptions() p = beam.Pipeline(options=pipeline_options) pipeline_options.view_as(SetupOptions).save_main_session = True pipeline_options.view_as(StandardOptions).streaming = True

क्योंकि Apache Beam Pub / Sub IO कनेक्टर का उपयोग केवल स्ट्रीमिंग मोड में आवश्यक है
पाइपलाइनों को जोड़ें () (हालांकि वास्तव में विकल्पों का उपयोग नहीं किया जाता है), अन्यथा, एक टेम्पलेट बनाएं
अपवाद के साथ गिरता है। यह टेम्पलेट लॉन्च करने के विकल्पों के बारे में कहा जाना चाहिए। वे हो सकते हैं
स्थिर और तथाकथित "रनटाइम"। यहाँ इस विषय पर प्रलेखन के लिए एक कड़ी है। विकल्प आपको अग्रिम में मापदंडों को निर्दिष्ट किए बिना एक टेम्पलेट बनाने की अनुमति देते हैं, लेकिन जब आप टेम्पलेट से डेटाफ़्लो जॉब शुरू करते हैं तो उन्हें पास करते हैं, लेकिन मैं अभी भी इसे लागू नहीं कर सका, शायद इस तथ्य के कारण कि यह कनेक्टर RuntimeValueProvider समर्थन नहीं करता है।

 # Read from PubSub into a PCollection. input_stream = p | beam.io.ReadFromPubSub(input_topic)

टिप्पणी से सब कुछ स्पष्ट है, हम विषय से धागा पढ़ते हैं। यह जोड़ने योग्य है कि आप स्ट्रीम ले सकते हैं
विषय और सदस्यता (सब्सक्रिप्शन) दोनों से। यदि विषय इनपुट के रूप में निर्दिष्ट है, तो
इस विषय के लिए एक अस्थायी सदस्यता स्वचालित रूप से बनाई जाएगी। वाक्यविन्यास भी सुंदर है
स्पष्ट, इनपुट डेटा स्ट्रीम beam.io.ReadFromPubSub(input_topic) हमारे लिए भेजा जाता है
पाइपलाइन p ।

 # Transform stream to BigQuery IO format stream_bq = input_stream | 'transform to BigQuery' >> beam.ParDo(TransformToBigQuery())

यह वह जगह है जहां ट्रांसफ़ॉर्म # 1 होता है और हमारे इनपुट को अजगर स्ट्रिंग से में बदल दिया जाता है
अजगर तानाशाह, और आउटपुट में हमें PCollection # 1 मिलता है। >> सिंटैक्स में दिखाई देता है। पर
वास्तव में, उद्धरण चिह्नों में पाठ धारा का नाम है (अद्वितीय होना चाहिए), साथ ही एक टिप्पणी,
जिसे GCP डेटाफ़्लो वेब इंटरफ़ेस में ग्राफ़ पर ब्लॉक में जोड़ा जाएगा। आइए अधिक विस्तार से विचार करें
ओवरराइड किए गए वर्ग TransformToBigQuery ।

 class TransformToBigQuery(beam.DoFn): #          ,   # BigQuery IO     python dict def process(self, element, *args, **kwargs): body = json.loads(element) #       ,      # python dict       ,     #  ,      python dict yield body

element चर में PubSub सदस्यता का एक संदेश होगा। जैसा कि देखा गया है
कोड, हमारे मामले में यह वैध JSON होना चाहिए। कक्षा में होना चाहिए
process विधि को फिर से परिभाषित किया गया है, जिसमें आवश्यक परिवर्तन किए जाने चाहिए
इनपुट लाइन आउटपुट से मिलती है जो सर्किट से मेल खाती है
वह तालिका जिसमें यह डेटा लोड किया जाएगा। क्योंकि इस मामले में हमारा प्रवाह है
अपाचे बीम के संदर्भ में निरंतर, unbounded , आपको इसका उपयोग करके वापस करना होगा
yield , अंतिम डेटा स्ट्रीम के लिए return नहीं। अंतिम प्रवाह के मामले में,
(और आवश्यक) इसके अलावा windowing और triggers कॉन्फ़िगर करें

 # Tag stream by schema name tagged_stream = \ stream_bq \ | 'tag data by type' >> beam.ParDo(TagDataWithReqType()).with_outputs(*schema_dct.keys(), main='default')

यह कोड PCollection # 1 को # 2 ट्रांसफ़ॉर्म करने का निर्देश देता है जहां टैगिंग होगी
(पृथक्करण) डेटा स्ट्रीम का। चर schema_dct इस मामले में, एक शब्दकोश, जहां कुंजी बिना विस्तार के स्कीम फ़ाइल का नाम है, यह टैग होगा, और मूल्य योजना का वैध JSON है
इस टैग के लिए BigQuery टेबल। यह ध्यान दिया जाना चाहिए कि इस योजना को वास्तव में प्रेषित किया जाना चाहिए
दृश्य {'fields': } जहाँ , JSON फॉर्म में BigQuery तालिका का स्कीमा है (आप कर सकते हैं)
वेब इंटरफेस से निर्यात)।

main='default' थ्रेड टैग का नाम है जिसके लिए वे जाएंगे
सभी संदेश जो टैगिंग शर्तों के अधीन नहीं हैं। वर्ग पर विचार करें
TagDataWithReqType ।

 class TagDataWithReqType(beam.DoFn): #      , ..      #     ,       #  with_outputs + default def process(self, element, *args, **kwargs): req_type = element.get('_') types = ( 'type1', 'type2', 'type3', ) if req_type in types: yield TaggedOutput(req_type, element) else: yield element

जैसा कि आप देख सकते हैं, यहां process वर्ग भी ओवरराइड है। types चर में नाम होते हैं
टैग और उन्हें संख्या और शब्द कुंजी के नाम के साथ संख्या और नाम से मेल खाना चाहिए
schema_dct । हालाँकि process विधि में तर्क स्वीकार करने की क्षमता है, मैं कभी नहीं
मैं उन्हें पास करने में सक्षम था। मुझे अब तक इसका कारण पता नहीं चला है।

आउटपुट पर, हमें टैग की संख्या में थ्रेड्स की एक टुकड़ी मिलती है, अर्थात् हमारी संख्या
पूर्वनिर्धारित टैग + डिफ़ॉल्ट थ्रेड जो टैग करने में विफल रहा।

 # Stream unidentified data to default table tagged_stream.default | 'push to default table' >> beam.io.WriteToBigQuery( '%s:%s.default' % ( gcp_project, bq_dataset, ), schema=parse_table_schema_from_json(schema_dct.get('default')), )

ट्रांसफ़ॉर्म # ... (वास्तव में, यह आरेख पर नहीं है, यह "शाखा" है) - हम डिफ़ॉल्ट स्ट्रीम लिखते हैं
डिफ़ॉल्ट तालिका के लिए।

tagged_stream.default - default टैग के साथ एक स्ट्रीम लिया जाता है, एक वैकल्पिक वाक्यविन्यास टैग किया जाता tagged_stream['default']

schema=parse_table_schema_from_json(schema_dct.get('default')) - यहाँ इस योजना को परिभाषित किया गया है
तालिका। कृपया ध्यान दें कि default.json फ़ाइल मान्य BigQuery तालिका स्कीमा के साथ है
वर्तमान schema_dir = './' निर्देशिका में होना चाहिए।

स्ट्रीम default नामक तालिका पर जाएगी।

यदि इस नाम की तालिका (इस परियोजना के दिए गए डेटासेट में) मौजूद नहीं है, तो यह
डिफ़ॉल्ट सेटिंग के लिए धन्यवाद योजना से स्वचालित रूप से बनाया जाएगा
create_disposition=BigQueryDisposition.CREATE_IF_NEEDED

 # Stream data to BigQuery tables by number of schema names for name, schema in schema_dct.iteritems(): tagged_stream[name] | 'push to table %s' % name >> beam.io.WriteToBigQuery( '%s:%s.%s' % ( gcp_project, bq_dataset, name), schema=parse_table_schema_from_json(schema), )

ट्रांसफ़र # 3, सब कुछ उन लोगों के लिए स्पष्ट होना चाहिए जो बहुत शुरुआत और खुद से लेख पढ़ते हैं
अजगर वाक्यविन्यास। हम एक लूप के साथ धारा को अलग करते हैं और प्रत्येक स्ट्रीम को अपनी तालिका में लिखते हैं
उसकी योजना। यह याद किया जाना चाहिए कि स्ट्रीम नाम अद्वितीय होना चाहिए - '%s:%s.%s' % (gcp_project, bq_dataset, name) ।

अब यह स्पष्ट होना चाहिए कि यह कैसे काम करता है और आप एक टेम्पलेट बना सकते हैं। इसके लिए आपको चाहिए
कंसोल में चलाएं (उपलब्ध होने पर वेन को सक्रिय करना न भूलें) या आईडीई से:

 python _.py / --runner DataflowRunner / --project dreamdata-test / --staging_location gs://STORAGE_NAME/STAGING_DIR / --temp_location gs://STORAGE_NAME/TEMP_DIR / --template_location gs://STORAGE_NAME/TEMPLATES_DIR/TEMPLATE_NAME

उसी समय, Google खाते तक पहुंच का आयोजन किया जाना चाहिए, उदाहरण के लिए, निर्यात के माध्यम से
GOOGLE_APPLICATION_CREDENTIALS पर्यावरण GOOGLE_APPLICATION_CREDENTIALS या अन्य तरीके से ।

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

यदि कोई त्रुटि नहीं हुई है, तो gs://STORAGE_NAME/TEMPLATES_DIR/TEMPLATE_NAME होगा
बनाया टेम्पलेट। यह कहने योग्य है कि gs://STORAGE_NAME/STAGING_DIR में भी लिखा जाएगा
सर्विस फाइल्स जो कि आधार पर बनाई गई डाटाफॉ जॉब के सफल संचालन के लिए आवश्यक हैं
टेम्पलेट और आपको उन्हें हटाने की आवश्यकता नहीं है।

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

4. निष्कर्ष

इस प्रकार, हम PubSub से BigQuery का उपयोग करके स्ट्रीम को स्ट्रीम करने में कामयाब रहे
आगे भंडारण, परिवर्तन और के उद्देश्य के लिए आवश्यक डेटा परिवर्तन
डेटा का उपयोग।

मुख्य लिंक

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