Data Agent Kit और Antigravity IDE की मदद से, धोखाधड़ी का पता लगाने वाली पाइपलाइन

1. परिचय

मान लें कि आप Cymbal Financial में डेटा साइंटिस्ट हैं. यह कंपनी, पेमेंट प्रोसेस करने वाली एक बड़ी कंपनी है. पेमेंट में देरी की कई शिकायतें मिली हैं. अनुपालन टीम को लगता है कि यह धोखाधड़ी एक साथ की गई है. आपको रॉ क्लियरिंगहाउस ट्रांज़ैक्शन लॉग को प्रोसेस करने के लिए एक पाइपलाइन बनानी होगी. साथ ही, डेटा को साफ़ करना होगा, मशीन लर्निंग मॉडल को ट्रेन करना होगा, बैच इन्फ़रेंस चलाना होगा, और ज़्यादा जोखिम वाले लेन-देन को मैन्युअल ऑडिट के लिए, Cloud Spanner की समीक्षा कतार में डालना होगा.

आम तौर पर, इसके लिए सेटअप कोड (स्पार्क नोटबुक, डीबीटी कॉन्फ़िगरेशन, ट्रेनिंग स्क्रिप्ट, Airflow DAG) को बार-बार लिखना पड़ता है. साथ ही, कंसोल इंटरफ़ेस और एडिटर के बीच लगातार स्विच करना पड़ता है.

इस कोडलैब में, Antigravity IDE में Google Cloud Data Agent Kit (DAK) का इस्तेमाल करके, किसी एजेंट के साथ पेयर-प्रोग्रामिंग की जाएगी. बातचीत वाली नैचुरल लैंग्वेज का इस्तेमाल करके, एजेंट आपकी इन कामों में मदद करेगा: स्पार्क नोटबुक जनरेट करना, dbt प्रोजेक्ट को कंपाइल करना, इन्फ़रेंस लूप बनाना, और Managed Service for Apache Airflow का इस्तेमाल करके वर्कफ़्लो को व्यवस्थित करना.

आपको क्या करना होगा

• Managed Service for Apache Spark (Spark Serverless) का इस्तेमाल करके, Cloud Storage से क्लियरिंगहाउस लॉग इंपोर्ट करें और उन्हें BigQuery टेबल में सेव करें.
• dbt का इस्तेमाल करके, लेन-देन को डुप्लीकेट होने से बचाएं और उन्हें नॉर्मलाइज़ करें. इससे साफ़ डेटा लेयर (रॉ, स्टेजिंग, बेहतर) तैयार की जा सकती हैं.
• Spark Serverless पर, डिस्ट्रीब्यूटेड रैंडम फ़ॉरेस्ट क्लासिफ़िकेशन मॉडल को ट्रेन करें (RandomForestClassifier).
• नए लेन-देन पर बैच इन्फ़रेंस की सुविधा चालू करें और ज़्यादा जोखिम वाली चेतावनियों को सीधे Cloud Spanner में लिखें.
• Managed Service for Apache Airflow का इस्तेमाल करके, पूरी पाइपलाइन को व्यवस्थित करें, उसे विज़ुअली कॉन्फ़िगर करें, और डिप्लॉय करें. साथ ही, IDE में इंटरैक्टिव DAG मॉनिटरिंग की सुविधा का इस्तेमाल करें.

आपको किन चीज़ों की ज़रूरत होगी

• कोई वेब ब्राउज़र, जैसे कि Chrome
• बिलिंग की सुविधा वाला Google Cloud प्रोजेक्ट. हमारा सुझाव है कि हैंड्स-ऑन लैब के लिए, एक नया प्रोजेक्ट इस्तेमाल करें.
• एसक्यूएल, Python, और PySpark की बुनियादी जानकारी.
• Google AI Pro की सदस्यता के साथ Antigravity IDE (सुझाया गया)

इस कोडलैब में बनाए गए संसाधनों की लागत 5 डॉलर से कम होनी चाहिए. प्रोविज़न किए गए संसाधनों को मिटाने के लिए, लैब के आखिर में दिए गए साफ़ करें निर्देशों का पालन ज़रूर करें.

अपना टूल चुनें: इस लैब में Antigravity IDE का इस्तेमाल किया गया है. यह एक्सटेंशन, VS Code के सभी फ़ोर्क पर काम करता है. VS Code के लिए अन्य वर्शन के साथ-साथ, Antigravity 2.0, Antigravity CLI, और Claude Code के लिए प्लग-इन पर आधारित वर्कफ़्लो जल्द ही उपलब्ध होंगे!

2. एनवायरमेंट सेटअप करना

लैब शुरू करने के लिए, आपको बूटस्ट्रैप स्क्रिप्ट चलानी होगी. यह स्क्रिप्ट, ज़रूरी GCP API अपने-आप चालू करती है. साथ ही, डेटा को इकट्ठा करने के लिए Cloud Storage बकेट बनाती है, नकली लेन-देन और डायरेक्ट्री डेटासेट जनरेट करती है, रेफ़रंस डायरेक्ट्री को BigQuery में लोड करती है, और Cloud Spanner और Managed Service for Apache Airflow (जिसे पहले Cloud Composer कहा जाता था) की बैकग्राउंड प्रोविज़निंग शुरू करती है.

कोई प्रोजेक्ट चुनें या बनाएं

Google Cloud Console में, कोई मौजूदा प्रोजेक्ट चुनें या नया प्रोजेक्ट बनाएं.

बिलिंग की पुष्टि करना

पक्का करें कि आपके Google Cloud प्रोजेक्ट के लिए बिलिंग की सुविधा चालू हो. ऐसा करने के तरीके के बारे में ज़्यादा जानने के लिए, यह गाइड पढ़ें.

सेटअप स्क्रिप्ट चलाना

एनवायरमेंट सेटअप करने के लिए, Google Cloud Shell या Google Cloud CLI के साथ कॉन्फ़िगर किए गए अपने लोकल शेल का इस्तेमाल करें.

1. Google Cloud Console खोलें.
2. सबसे ऊपर दाएं कोने में मौजूद टूलबार में, Cloud Shell चालू करें पर क्लिक करें.

3. Cloud Shell टर्मिनल में, अपना ऐक्टिव प्रोजेक्ट कॉन्फ़िगर करें:

gcloud config set project <<YOUR_PROJECT_ID>>
export PROJECT_ID=$(gcloud config get-value project)

4. codelab रिपॉज़िटरी को क्लोन करें और स्क्रिप्ट फ़ोल्डर पर जाएं:

cd ~/
git clone --filter=blob:none --no-checkout https://github.com/GoogleCloudPlatform/devrel-demos.git
cd ~/devrel-demos
git sparse-checkout init --cone
git sparse-checkout set codelabs/agentic-data-labs/data-science
git checkout main
cd codelabs/agentic-data-labs/data-science/scripts

5. सभी संसाधनों को us-central1 में डिप्लॉय करने के लिए, बूटस्ट्रैप सेटअप स्क्रिप्ट चलाएं:

chmod +x setup.sh setup_spanner.sh setup_composer.sh
export REGION=us-central1
./setup.sh

setup.sh क्या करता है:

• एपीआई चालू करता है: BigQuery, Dataproc, Spanner, मैनेज किया गया Airflow, Cloud Storage, और बुनियादी ढांचे से जुड़े एपीआई चालू करता है.
• Cloud Storage बकेट बनाता है: यह रॉ ट्रांज़ैक्शन लॉग और एमएल मॉडल आर्टफ़ैक्ट के लिए बकेट उपलब्ध कराता है.
• शुरुआती डेटा को स्टेज करता है: रॉ लेन-देन के लॉग को Cloud Storage में कॉपी करता है और रेफ़रंस डेटा को BigQuery में लोड करता है.
• Spark Serverless को कॉन्फ़िगर करता है: Cloud Spanner कनेक्टर के साथ रनटाइम सेशन टेंप्लेट बनाता है.
• बैकग्राउंड सेवाएं उपलब्ध कराता है: Cloud Spanner और Managed Airflow बनाने के लिए, बैकग्राउंड वर्कर शुरू करता है.
• जनरेट करता है .env: यह कोडलैब और टियरडाउन स्क्रिप्ट के लिए एनवायरमेंट कॉन्फ़िगरेशन सेव करता है.

सेटअप स्क्रिप्ट से बनाई गई ऐसेट, ./teardown.sh के साथ फ़ाइनल क्लीन अप चरण में मिटा दी जाती हैं.

6. स्क्रिप्ट पूरी होने के बाद, आपको एक खास आउटपुट दिखेगा. इससे पता चलेगा कि आपका BigQuery डेटासेट और Cloud Storage बकेट तैयार है. बैकग्राउंड में, Cloud Spanner (इसमें ~2 मिनट लगते हैं) और Managed Airflow (इसमें ~20 मिनट लगते हैं) का प्रावधान जारी रहेगा. इनकी प्रोग्रेस को किसी भी समय मॉनिटर किया जा सकता है. इसके लिए, यह कमांड चलाएं:

tail -f /tmp/spanner_setup.log
tail -f /tmp/composer_setup.log

ऑप्टिमाइज़ किया गया डेवलपमेंट लूप: Managed Airflow को चालू होने में करीब 20 मिनट लगते हैं. इसलिए, सेटअप स्क्रिप्ट इसे तुरंत बैकग्राउंड में चालू कर देती है. इससे Airflow के चालू होने के दौरान, पेज 3 (डेटा इंटेक), पेज 4 (dbt ट्रांसफ़ॉर्मेशन), पेज 5 (मशीन लर्निंग ट्रेनिंग), और पेज 6 (बैच इन्फ़रेंस) पर काम किया जा सकता है. पेज 7 (ऑर्केस्ट्रेशन) पर पहुंचने तक, आपका एनवायरमेंट पूरी तरह से तैयार हो जाएगा.

Antigravity IDE खोलें

1. Google Antigravity के डाउनलोड पेज से, Antigravity IDE को डाउनलोड और इंस्टॉल करें.
2. Antigravity IDE लॉन्च करें.
3. अपनी लोकल मशीन पर एक नया, खाली फ़ोल्डर बनाएं. उदाहरण के लिए, agentic-data-labs नाम का फ़ोल्डर बनाएं. इसके बाद, फ़ोल्डर खोलें को चुनकर, उसे IDE में खोलें. यह कोडलैब के लिए, आपके लोकल वर्कस्पेस के तौर पर काम करेगा.

Data Agent Kit एक्सटेंशन इंस्टॉल करना

Google Cloud Data Agent Kit एक्सटेंशन, Google Cloud की डेटा सेवाओं के साथ सीधे तौर पर इंटिग्रेट होता है. इससे आपको अपने एडिटर में ही BigQuery, Cloud SQL, Cloud Storage वगैरह के साथ इंटरैक्ट करने की सुविधा मिलती है. इसके लिए, आपको कॉन्टेक्स्ट स्विच करने की ज़रूरत नहीं होती.

1. Antigravity IDE में, स्क्रीन की बाईं ओर मौजूद Activity Bar में, Extensions आइकॉन पर क्लिक करें. यह आइकॉन, चार स्क्वेयर की तरह दिखता है.
2. एक्सटेंशन पैनल में सबसे ऊपर मौजूद खोज बार में, Google Cloud Data Agent Kit टाइप करें.
3. googlecloudtools ने Google Cloud Data Agent Kit नाम का एक्सटेंशन पब्लिश किया है. इसे ढूंढें
4. इंस्टॉल करें बटन पर क्लिक करें.
5. आपको यह मैसेज दिख सकता है, "क्या आपको पब्लिशर ‘googlecloudtools' और उसके एक्सटेंशन पर भरोसा है?". आगे बढ़ने के लिए, भरोसेमंद पब्लिशर और इंस्टॉल करें पर क्लिक करें.

इंस्टॉल हो जाने के बाद, आपको Antigravity IDE के सबसे बाईं ओर मौजूद ऐक्टिविटी बार में, नया Google Cloud Data Agent Kit आइकॉन दिखेगा.

6. "Google Cloud Data Agent Kit में आपका स्वागत है" नाम का एक पेज अपने-आप खुल जाएगा. अगर आपने Cloud खाते में साइन इन नहीं किया है, तो ऐक्सेस की अनुमति देने के लिए दिए गए निर्देशों का पालन करें.
7. कॉन्फ़िगरेशन की खास जानकारी सेक्शन में, प्रोजेक्ट फ़ील्ड ढूंढें. ड्रॉपडाउन पर क्लिक करें और अपना Google Cloud प्रोजेक्ट चुनें. अपने देश/इलाके को us-central1 के तौर पर सेट करें. इसके बाद, Configure MCP Servers को चुनें.

8. एमसीपी सर्वर कॉन्फ़िगर करें को चुनें. एमसीपी कॉन्फ़िगरेशन पैनल में जाकर, पक्का करें कि आपने इन रिमोट एमसीपी सर्वर को चालू किया हो:
   • BigQuery
   • Spanner
   • नोटबुक

इसके बाद, शुरू करें पर क्लिक करें.

क्या आपने पहले ही इंस्टॉल कर लिया है या वेलकम पेज नहीं खुला? एक्सटेंशन की सेटिंग को कभी भी मैन्युअल तरीके से खोला जा सकता है. अपने कॉन्फ़िगरेशन को टॉगल करने के लिए, सबसे बाईं ओर मौजूद ऐक्टिविटी बार में, Google Cloud Data Agent Kit आइकॉन पर क्लिक करें. इसके बाद, सेटिंग चुनें.

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

कॉन्फ़िगरेशन के विकल्प देखना

सेटअप पूरा होने के बाद, आपको "Google Cloud Data Agent Kit का इस्तेमाल शुरू करें" पेज पर रीडायरेक्ट कर दिया जाएगा.

1. "सेटअप और कॉन्फ़िगरेशन" में जाकर, शुरू करें पर क्लिक करें.
2. इससे डेटा एजेंट किट का कॉन्फ़िगरेशन पैनल खुलता है. टैब एक्सप्लोर करें:
   • प्रोजेक्ट और क्षेत्र: चुने गए प्रोजेक्ट आईडी की पुष्टि करें. साथ ही, यह पक्का करें कि सेटअप स्क्रिप्ट ने सभी ज़रूरी एपीआई (Compute Engine, Cloud Storage, BigQuery, Spanner वगैरह) चालू किए हों.
   • BigQuery: BigQuery क्वेरी के लिए जगह की डिफ़ॉल्ट जानकारी कॉन्फ़िगर करें. क्षेत्र us-central1 का इस्तेमाल करें.
   • एमसीपी सर्वर कॉन्फ़िगर करें: चालू किए गए एमसीपी सर्वर (BigQuery, Notebooks, Spanner वगैरह) देखें. इनकी मदद से, एआई एजेंट आपके डेटा के साथ सुरक्षित तरीके से इंटरैक्ट कर सकते हैं.
   • स्किल: पहले से बनी हुई स्किल एक्सप्लोर करें. ये स्किल, एजेंट को मुश्किल डेटा टास्क के लिए खास सुविधाएं देती हैं.

सेक्शन की खास जानकारी: आपने GCS और BigQuery ऐसेट बनाने के लिए, बूटस्ट्रैप स्क्रिप्ट चलाई. वहीं, Spanner और Airflow बैकग्राउंड में काम करते हैं. इसके बाद, आपने Antigravity IDE में प्रोजेक्ट खोला और Google Cloud Data Agent Kit एक्सटेंशन चालू किया. अब आपके पास अपनी पहली नोटबुक लिखने का विकल्प है.

3. Spark Serverless का इस्तेमाल करके, रॉ लॉग को प्रोसेस करना

इस सेक्शन में, डेटा लेक में रॉ JSON ट्रांज़ैक्शन लॉग डाले जाएंगे. Apache Spark के लिए मैनेज की गई सेवा (Spark Serverless), BigQuery के नेटिव स्टोरेज से सीधे कनेक्ट होती है. टेबल के फ़ॉर्मैट में मौजूद डेटा को मैनेज करने के लिए, BigQuery के स्टैंडर्ड कनेक्टर का इस्तेमाल किया जाएगा. इससे सीधे तौर पर क्वेरी करने और डेटा का विश्लेषण करने की सुविधा मिलेगी.

पहले से कॉन्फ़िगर किए गए Spark Serverless रनटाइम के बारे में जानें

स्पार्क कोड को चलाने से पहले, सेटअप स्क्रिप्ट से पहले से कॉन्फ़िगर किए गए Serverless Runtime टेंप्लेट की जांच करें. यह टेंप्लेट, टारगेट एक्ज़ीक्यूशन एनवायरमेंट के बैकएंड को तय करता है. साथ ही, कनेक्टर की ज़रूरी डिपेंडेंसी को बंडल करता है.

1. IDE की गतिविधि बार में, Google Cloud Data Agent Kit पैनल खोलें.
2. Apache Spark ड्रॉप-डाउन मेन्यू को बड़ा करें. इसके बाद, Serverless को बड़ा करें.
3. fraud-pipeline-runtime पर राइट क्लिक करें और प्रोफ़ाइल चुनें. इससे एडिटर में, प्रोफ़ाइल का कॉन्फ़िगरेशन व्यू खुल जाएगा.
4. प्रोफ़ाइल टैब में, नीचे की ओर स्क्रोल करें और प्रॉपर्टी को बड़ा करें. इससे एनवायरमेंट से जुड़ी कस्टम डिपेंडेंसी की जांच की जा सकती है:
   • spark.jars: इसमें gs://spark-lib/spanner/spark-3.5-spanner-1.4.0.jar शामिल है. यह Spark Spanner कनेक्टर का इस्तेमाल करता है, ताकि Spark के जॉब, अनुमान के नतीजों को सीधे Cloud Spanner में लिख सकें. ऐसा लैब में बाद में किया जा सकता है. (ध्यान दें: Dataproc Serverless में, Google Cloud का Spark BigQuery कनेक्टर डिफ़ॉल्ट रूप से शामिल होता है. इसलिए, BigQuery टेबल को पढ़ने और लिखने के लिए, किसी अतिरिक्त जार कॉन्फ़िगरेशन की ज़रूरत नहीं होती).

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

Data Agent Kit का इस्तेमाल करके डेटा इनजेस्ट करना

आपको Spark सेशन को मैन्युअल तरीके से कॉन्फ़िगर करने या PySpark की लोडिंग स्क्रिप्ट को शुरू से लिखने के बजाय, डेटा एजेंट किट का इस्तेमाल करके किसी एजेंट के साथ पेयर-प्रोग्रामिंग करनी होगी.

1. सबसे ऊपर दाएं कोने में मौजूद टूलबार में, टॉगल एजेंट आइकॉन पर क्लिक करके, एजेंट चैट पैनल खोलें.
2. नीचे दिए गए प्रॉम्प्ट को चैट में चिपकाएं. ${PROJECT_ID} की जगह Google Cloud का अपना प्रोजेक्ट आईडी डालना न भूलें:

Create a PySpark notebook (01_ingestion.ipynb) to ingest JSON transaction logs
from gs://${PROJECT_ID}-fin-clearing-raw/ into a BigQuery table
`${PROJECT_ID}.transactions_dataset_evals.raw_transactions`
using the Spark BigQuery connector (`format("bigquery")`) with overwrite mode.

3. अगर एजेंट, बैकग्राउंड की पुष्टि करने वाली कमांड (जैसे, "क्या इस कमांड को चलाने की अनुमति देनी है?") को लागू करने की अनुमति मांगता है, तो सुझाई गई कमांड की समीक्षा करें. इसके बाद, हां, इस बार अनुमति दें या हां, हमेशा अनुमति दें को चुनें.
4. जब एजेंट फ़ाइल जनरेट कर लेता है, तब चैट पैनल में सबसे नीचे मौजूद नीले रंग के सभी स्वीकार करें बटन (या सही का निशान वाला आइकॉन) पर क्लिक करके, notebooks/01_ingestion.ipynb को अपने वर्कस्पेस में सेव करें.

कमांड को अनुमति देना और बदलावों को स्वीकार करना: एआई, पेयर-प्रोग्रामर के तौर पर काम करता है. इसलिए, कोड जनरेट करने से पहले, एजेंट सुरक्षित टर्मिनल कमांड (जैसे, सिंटैक्स की जांच या डिपेंडेंसी की जांच) चलाने की अनुमति मांग सकता है. जब आपसे कहा जाए, तब इन कमांड की समीक्षा करें और इन्हें अनुमति दें. साथ ही, जब भी फ़ाइलें या लागू करने के प्लान दिखाए जाएं, तब चैट पैन में आगे बढ़ें या सभी स्वीकार करें पर क्लिक करें.

नोटबुक की समीक्षा करना और उसे लागू करना

1. आईडीई में, हाल ही में जनरेट किया गया notebooks/01_ingestion.ipynb खोलें.
2. BigQuery कनेक्टर के राइट लॉजिक के लिए, PySpark कोड की समीक्षा करें.
3. IDE के नोटबुक टूलबार में, सभी सेल चलाएं पर क्लिक करें.
4. अगर रिमोट Spark नोटबुक को पहली बार चलाया जा रहा है, तो IDE आपसे लोकल डिपेंडेंसी इंस्टॉल करने के लिए कह सकता है. अगर कहा जाए, तो रिमोट स्पार्क कर्नल के लिए डिपेंडेंसी इंस्टॉल करें पर क्लिक करें. इसके बाद, इंस्टॉलेशन डायलॉग की पुष्टि करें. इसके बाद, सभी चलाएं पर फिर से क्लिक करें.
5. Select Kernel ड्रॉपडाउन मेन्यू में, Remote Spark Kernels -> fraud-pipeline-runtime on Serverless Spark चुनें. (अहम जानकारी: अगर आपको पहले से कॉन्फ़िगर किया गया रनटाइम टेंप्लेट नहीं दिखता है, तो कर्नल पिकर ड्रॉपडाउन में सबसे ऊपर दाईं ओर मौजूद रीफ़्रेश आइकॉन पर क्लिक करें, ताकि उपलब्ध रिमोट कर्नल फिर से लोड हो जाएं).
6. एडिटर में सबसे नीचे बाईं ओर मौजूद स्टेटस बार देखें. आपको Connecting to kernel: fraud-pipeline-runtime on Serverless Spark... दिखेगा. Spark Serverless runtime kernel backend को पहली बार लॉन्च किया जा रहा है. इसलिए, इसे चालू होने में कुछ मिनट लगेंगे.
7. कर्नल के कनेक्ट होने के बाद, नोटबुक सभी सेल को क्रम से अपने-आप एक्ज़ीक्यूट करना शुरू कर देगी. इससे, तैयार नहीं किए गए लेन-देन के लॉग को आपके BigQuery डेटासेट में प्रोसेस किया जा सकेगा.

Spark Serverless से कनेक्ट करना: Spark Serverless रनटाइम सेशन से कनेक्ट करने पर, आपको स्टेटस बार में सबसे नीचे Connecting to kernel... दिखेगा. नोटबुक के कनेक्ट होने और सभी सेल को एक्ज़ीक्यूट करने से पहले, Google Cloud पर संसाधनों को शुरू होने में करीब तीन मिनट लगते हैं.

पुष्टि

एक्ज़ीक्यूशन पूरा होने के बाद, टेबल बनने की पुष्टि करने के लिए, Data Agent Kit कैटलॉग देखें:

1. IDE की गतिविधि बार में, Google Cloud Data Agent Kit पैनल खोलें.
2. कैटलॉग सेक्शन को बड़ा करें.
3. अपना प्रोजेक्ट आईडी बड़ा करें.
4. BigQuery को बड़ा करें.
5. transactions_dataset_evals डेटासेट को बड़ा करें.
6. मुख्य एडिटर में टेबल की ज़्यादा जानकारी वाला व्यू खोलने के लिए, raw_transactions टेबल पर क्लिक करें.
7. बाईं ओर मौजूद नेविगेशन में, डेटा, स्कीमा, और जानकारी टैब एक्सप्लोर करें. इससे, आपको शामिल किए गए रिकॉर्ड और मेटाडेटा की जांच करने में मदद मिलेगी.

सेक्शन की खास जानकारी: आपने एजेंट चैट में नैचुरल लैंग्वेज का इस्तेमाल करके, Spark Serverless के पूरे वर्कलोड को जनरेट किया. इसके बाद, आपने इसका इस्तेमाल करके अनस्ट्रक्चर्ड JSON लॉग को BigQuery (रॉ) टेबल में प्रोसेस किया.

4. dbt की मदद से डुप्लीकेट कॉपी हटाना और डेटा को सामान्य बनाना

एमएल मॉडल को ट्रेनिंग देने से पहले, डेटा क्वालिटी को बेहतर बनाया जाएगा. इसके लिए, डुप्लीकेट स्ट्रीमिंग लॉग हटाए जाएंगे, खराब रिकॉर्ड (जैसे कि खाली लेन-देन आईडी) अलग किए जाएंगे, और डाइमेंशनल डेटा (पेयर और पेमेंट पाने वाला व्यक्ति या कारोबार) को जोड़ा जाएगा. इस प्रोसेस के लिए, आइडमपोटेंट और भरोसेमंद एसक्यूएल ट्रांसफ़ॉर्मेशन की ज़रूरत होती है. इसलिए, dbt (डेटा बिल्ड टूल) एक बेहतरीन विकल्प है.

dbt पाइपलाइन का ढांचा तैयार करना

BigQuery डेटासेट पर dbt प्रोजेक्ट जनरेट करने के लिए, एजेंट का इस्तेमाल करें:

1. एजेंट चैट पैनल पर वापस जाएं.
2. dbt प्रोजेक्ट जनरेट करने के लिए, यह निर्देश दें:

Scaffold a us-central1 dbt project in dbt_project/ that maps raw_transactions
through to an enriched_transactions model in dataset transactions_dataset_evals.

Deduplicate by transaction_id in staging. Quarantine null IDs to an invalid_transactions model.
Join the valid staging records with dim_payers and dim_payees for the enriched_transactions model,
preserving the historical `is_fraud` label column, and finally add a transaction uniqueness test.

Create an implementation plan first.

3. एजेंट, मुख्य एडिटर पैन में लागू करने का प्लान आर्टफ़ैक्ट दिखाएगा. प्रस्तावित फ़ाइल स्ट्रक्चर और एसक्यूएल लॉजिक की समीक्षा करें.
4. एजेंट को आपके फ़ाइल फ़ोल्डर में फ़ाइलें जनरेट करने की अनुमति देने के लिए, आगे बढ़ें पर क्लिक करें. इसके बाद, सभी स्वीकार करें पर क्लिक करें.

5. जनरेट होने के बाद, एजेंट एक वॉकट्रू दिखाता है. इसमें नए कॉम्पोनेंट के बारे में खास जानकारी होती है. अगर कहा जाए, तो सभी बदलाव स्वीकार करें.

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

बनाना और टेस्ट करना

हालांकि, एजेंट ने जनरेट किए गए एसक्यूएल की सिंटैक्टिकल पुष्टि करने के लिए, dbt compile को अपने-आप चलाया था. अब आपको इन व्यू और टेबल को BigQuery में बदलना होगा. साथ ही, स्थानीय पुष्टि के लिए डेटा क्वालिटी टेस्ट चलाने होंगे. (ध्यान दें: लैब में बाद में, इस dbt चरण को एंड-टू-एंड Airflow DAG के हिस्से के तौर पर ऑटोमेट किया जाएगा).

1. सबसे बाईं ओर मौजूद गतिविधि बार में, एक्सप्लोरर आइकॉन पर क्लिक करें या Cmd/Ctrl+Shift+E दबाएं.
2. जनरेट किए गए एसक्यूएल मॉडल की जांच करने के लिए, dbt_project -> models को बड़ा करें. एडिटर में, डेटा में बदलाव करने और धोखाधड़ी रोकने से जुड़ी सुविधा के लॉजिक को खोलने और उसकी समीक्षा करने के लिए, enriched_transactions.sql पर क्लिक करें.
3. File Explorer में, dbt_project फ़ोल्डर पर राइट क्लिक करें और Integrated Terminal में खोलें को चुनें. इससे, टर्मिनल पैन अपने-आप खुल जाता है. इसे सीधे तौर पर ज़रूरी dbt_project वर्किंग डायरेक्ट्री पर सेट किया जाता है.
4. अगर आपने पहले से dbt इंस्टॉल नहीं किया है, तो dbt_project/ के बाहर (अपने घर या वर्कस्पेस के रूट पर) एक वर्चुअल एनवायरमेंट बनाएं और BigQuery अडैप्टर इंस्टॉल करें:

python3 -m venv ~/.venv/dbt
source ~/.venv/dbt/bin/activate
pip install dbt-bigquery

5. dbt मॉडल और उनसे जुड़े डेटा क्वालिटी टेस्ट चलाएं:

dbt build

6. टर्मिनल आउटपुट देखें. dbt, एसक्यूएल को कंपाइल करेगा. साथ ही, BigQuery में स्टेजिंग और बेहतर टेबल को मटीरियलाइज़ करेगा. इसके बाद, डेटा टेस्ट को एक्ज़ीक्यूट करेगा.

एजेंट की मदद से बिल्ड से जुड़ी समस्याओं को हल करना: अगर dbt build को कोई गड़बड़ी मिलती है (जैसे, स्कीमा का मेल न खाना या संबंध मौजूद न होना), तो गड़बड़ी के आउटपुट को सीधे एजेंट चैट में कॉपी करके चिपकाएं. एजेंट, गड़बड़ी का विश्लेषण करेगा, मॉडल में बदलाव करेगा, और आपके लिए समस्या हल करेगा.

7. बिल्ड पूरा होने के बाद, टर्मिनल पैन को बंद करें, ताकि बाकी चरणों के लिए स्क्रीन पर जगह खाली हो जाए.

सेक्शन की खास जानकारी: आपने एजेंट की मदद से dbt प्रोजेक्ट जनरेट किया, डेटा क्वालिटी टेस्ट किए, और रॉ रिकॉर्ड को स्टेजिंग और बेहतर BigQuery टेबल में बदला.

5. रैंडम फ़ॉरेस्ट की मदद से, धोखाधड़ी का पता लगाने वाले डिस्ट्रिब्यूटेड मॉडल को ट्रेन करना

BigQuery में मौजूद बेहतर बनाए गए लेन-देन के डेटा की मदद से, धोखाधड़ी वाले इवेंट का पता लगाने के लिए मशीन लर्निंग मॉडल बनाया जाएगा. रैंडम फ़ॉरेस्ट, ग्रुप लर्निंग का एक तरीका है. यह टेबल के फ़ॉर्मैट में मौजूद क्लासिफ़िकेशन डेटा के लिए सबसे सही है. Spark Serverless पर RandomForestClassifier चलाने से, मॉडल की ट्रेनिंग को वर्कर नोड में डिस्ट्रिब्यूट किया जाता है. इसके लिए, आपको इंफ़्रास्ट्रक्चर को मैनेज करने की ज़रूरत नहीं होती.

इस चरण में, Spark ML ट्रेनिंग पाइपलाइन जनरेट करने के लिए एजेंट का इस्तेमाल किया जाएगा.

एमएल ट्रेनिंग नोटबुक जनरेट करना

1. एजेंट से चैट करें पैनल खोलें.
2. मॉडल ट्रेनिंग के क्रम को डिज़ाइन करने के लिए, यह प्रॉम्प्ट दें. ${PROJECT_ID} की जगह अपने ऐक्टिव प्रोजेक्ट आईडी का इस्तेमाल करना न भूलें:

Create a PySpark notebook (02_training.ipynb) to train a distributed Random Forest
(RandomForestClassifier) model on the BigQuery table
`transactions_dataset_evals`.`enriched_transactions`, predicting the `is_fraud` label.
Train only on historically labeled records where `is_fraud` is not null.

One-hot encode categorical strings, scale amounts, cache the dataset in memory,
evaluate AUC, and save the evaluated model to gs://${PROJECT_ID}-models/fraud_model.

3. एजेंट के प्लान या जनरेट किए गए कोड की समीक्षा करें. इसके बाद,  को अपने वर्कस्पेस में सेव करने के लिए, आगे बढ़ें / सभी स्वीकार करें पर क्लिक करें.notebooks/02_training.ipynb

नोटबुक की समीक्षा करना और उसे लागू करना

1. एडिटर में notebooks/02_training.ipynb खोलें.
2. PySpark ML पाइपलाइन के चरणों की समीक्षा करें. इनमें फ़ीचर एन्कोडिंग, वेक्टर असेंबली, और रैंडम फ़ॉरेस्ट क्लासिफ़िकेशन लॉजिक शामिल हैं.
3. IDE के नोटबुक टूलबार में, सभी सेल चलाएं पर क्लिक करें.
4. Select Kernel ड्रॉपडाउन पिकर खुलने पर, fraud-pipeline-runtime on Serverless Spark को चुनें.

पुष्टि

मॉडल को लागू करने के बाद, पुष्टि करें कि उसे सही तरीके से ट्रेन किया गया है और एक्सपोर्ट किया गया है:

1. रिपोर्ट किए गए ROC (एयूसी) स्कोर की पुष्टि करने के लिए, नोटबुक में सबसे नीचे मौजूद मूल्यांकन सेल के आउटपुट देखें.
2. मॉडल आर्टफ़ैक्ट को GCS में सेव किया गया है या नहीं, यह देखने के लिए Data Agent Kit के साइडबार में मौजूद STORAGE एक्सप्लोरर पैन को बड़ा करें.
3. -models से खत्म होने वाले बकेट (आपके चालू प्रोजेक्ट आईडी से जुड़ा) को ढूंढें. इसके बाद, उसे बड़ा करें और यह देखने के लिए ड्रिल-डाउन करें कि fraud_model डायरेक्ट्री और उसके पाइपलाइन स्टेज मौजूद हैं या नहीं.

नोटबुक से प्रोडक्शन तक: इंटरैक्टिव नोटबुक, एक्सपेरिमेंट और मॉडल डेवलपमेंट के लिए बेहतरीन होती हैं. हालांकि, आपका एजेंट नोटबुक कोड को पैरामीटर वाली स्टैंडअलोन PySpark स्क्रिप्ट में भी बदल सकता है. इससे, Dataproc Serverless पर बार-बार बैच एक्ज़ीक्यूट किया जा सकता है.

सेक्शन की खास जानकारी: आपने एजेंट का इस्तेमाल करके, PySpark ML ट्रेनिंग पाइपलाइन बनाई. साथ ही, आपने अपनी बेहतर बनाई गई BigQuery टेबल पर रैंडम फ़ॉरेस्ट मॉडल को ट्रेन किया और मॉडल को Cloud Storage में एक्सपोर्ट किया.

6. बैच इन्फ़ेरेंस और Cloud Spanner में डेटा लिखना

Cloud Storage में सेव किए गए, ट्रेन किए गए अनुमानित मॉडल की मदद से, BigQuery से आने वाले नए लेन-देन पर बैच इन्फ़रेंस चलाया जाएगा. ज़्यादा जोखिम वाले लेन-देन को चालू सिस्टम पर रूट किया जाना चाहिए, ताकि अनुपालन टीम उनकी समीक्षा कर सके. Cloud Spanner, समीक्षा के लिए सबमिट किए गए आइटम की इस सूची के लिए, ज़रूरत के हिसाब से बढ़ाया जा सकने वाला ट्रांज़ैक्शनल डेटाबेस उपलब्ध कराता है.

बैच इन्फ़रेंस नोटबुक जनरेट करना

BigQuery, Cloud Storage, और Cloud Spanner को कनेक्ट करने वाला अनुमान नोटबुक बनाने के लिए, एजेंट का इस्तेमाल करें:

1. एजेंट से चैट करें पैनल खोलें.
2. यह प्रॉम्प्ट दें:

Create an inference notebook (03_inference.ipynb) that loads the RandomForestClassifier
model to score unlabeled records (where `is_fraud` is null) from the BigQuery table
`transactions_dataset_evals`.`enriched_transactions`.

Filter for high-risk transactions with a 50%+ fraud probability score (probability >= 0.50)
and write them to the Cloud Spanner table SparkEvalFraudReviewQueue in the cymbal-fraud instance
under fraud-db.

3. जनरेट की गई नोटबुक को स्वीकार करके, notebooks/03_inference.ipynb को अपने Workspace में सेव करें.

50% स्कोर को ज़्यादा जोखिम वाला क्यों माना जाता है: 50% स्कोर को सिक्के के उछाल की तरह माना जाता है. हालांकि, ज़्यादातर असली लेन-देन का स्कोर 5% से कम होता है. 50%से ज़्यादा स्कोर का मतलब है कि नेटवर्क में सबसे ज़्यादा अनियमितता वाले 2% लेन-देन. इससे नॉइज़ को फ़िल्टर किया जाता है, ताकि ऑडिट करने वाले लोग सिर्फ़ ज़्यादा जोखिम वाले असली इवेंट की समीक्षा कर सकें.

नोटबुक की समीक्षा करना और उसे लागू करना

1. एडिटर में, नई जनरेट की गई notebooks/03_inference.ipynb खोलें.
2. PySpark के अनुमान लगाने के क्रम की समीक्षा करें:
   • डिपेंडेंसी: Serverless Runtime टेंप्लेट, Spark को एक्ज़ीक्यूट करने के लिए ज़रूरी cloud-spanner JAR डिपेंडेंसी उपलब्ध कराता है.
   • डेटा फ़ॉर्मैट करना: स्पैनर टेबल स्कीमा से मैच करने के लिए, यह स्क्रिप्ट स्पार्क एमएल के जटिल वेक्टर कॉलम (जैसे कि रॉ फ़ीचर और संभावनाएं) को ड्रॉप करती है.
   • Spanner कनेक्टर: यह .format("cloud-spanner") का इस्तेमाल करके, फ़्लैग की गई लाइनों को लिखता है, ताकि उन्हें सीधे समीक्षा के लिए भेजी गई लाइनों की सूची में जोड़ा जा सके.
3. IDE के नोटबुक टूलबार में, सभी सेल चलाएं पर क्लिक करें.
4. जब आपसे कर्नल चुनने के लिए कहा जाए, तब fraud-pipeline-runtime on Serverless Spark चुनें.

पुष्टि

इन्फ़रेंस नोटबुक के प्रोसेस होने के बाद, सीधे तौर पर आईडीई में जाकर अपने ऑपरेशनल Spanner डेटाबेस से क्वेरी की जा सकती है:

1. IDE की गतिविधि बार में, Google Cloud Data Agent Kit पैनल खोलें.
2. कैटलॉग सेक्शन को बड़ा करें.
3. अपने प्रोजेक्ट आईडी को बड़ा करें. इसके बाद, Spanner को बड़ा करें.
4. cymbal-fraud -> fraud-db -> Tables -> SparkEvalFraudReviewQueue पर जाएं.
5. टेबल पर राइट क्लिक करके, क्वेरी टेबल चुनें. इसके बाद, क्वेरी चलाएं:

SELECT *
FROM `SparkEvalFraudReviewQueue`
LIMIT 100;

6. नीचे दिए गए क्वेरी के नतीजे वाले पैनल में, आपको नई जोड़ी गई ऐसी लाइनें दिखेंगी जिनमें ज़्यादा जोखिम वाले लेन-देन दिखाए गए हैं. इन लेन-देन को मैन्युअल तरीके से समीक्षा करने के लिए फ़्लैग किया गया है.

सेक्शन की खास जानकारी: आपने एजेंट का इस्तेमाल करके, बैच इन्फ़रेंस नोटबुक बनाई. साथ ही, आपने ट्रेनिंग दिए गए मॉडल की मदद से, बिना लेबल वाले BigQuery रिकॉर्ड को स्कोर किया. इसके अलावा, आपने ज़्यादा जोखिम वाले लेन-देन को सीधे तौर पर Cloud Spanner में लिखा.

7. Managed Airflow की मदद से, वर्कफ़्लो को व्यवस्थित और मैनेज करना

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

Managed Service for Apache Airflow (जिसे पहले Cloud Composer कहा जाता था) इस वर्कफ़्लो के लिए, मैनेज किया गया ऑर्केस्ट्रेशन इंजन उपलब्ध कराता है. Data Agent Kit में ऑर्केस्ट्रेशन पाइपलाइन की सुविधा शामिल होती है. यह सुविधा, डिक्लेरेटिव YAML पाइपलाइन की परिभाषाओं को सीधे तौर पर Airflow DAG में बदल देती है.

पाइपलाइन तय करना

ऑर्केस्ट्रेशन पाइपलाइन का कॉन्फ़िगरेशन जनरेट करने के लिए, एजेंट का इस्तेमाल करें:

1. एजेंट चैट में, यह प्रॉम्प्ट दें. ${PROJECT_ID} को अपनी जानकारी से बदलना न भूलें:

Initialize and define an orchestration pipeline (fraud_analysis_pipeline) triggering
the ingestion notebook, dbt project, and inference notebook in sequential order.
For Dataproc Serverless engine configs in us-central1, use resourceProfile.inline
(defining properties with spark.jars: "gs://spark-lib/spanner/spark-3.5-spanner-1.4.0.jar"
for inference) rather than resourceProfile.path or overrides.

Set the schedule interval to run daily at midnight, and use
gs://${PROJECT_ID}-airflow-artifacts for artifact storage.

DAG कॉन्फ़िगरेशन की समीक्षा करना

Data Agent Kit Orchestrator, डिक्लेरेटिव YAML कॉन्फ़िगरेशन का इस्तेमाल करके, Apache Airflow में पाइपलाइन तय करता है और उन्हें डिप्लॉय करता है. इससे, वर्शन कंट्रोल की मदद से परिभाषाएं तय की जा सकती हैं और CI/CD के ज़रिए उन्हें डिप्लॉय किया जा सकता है.

IDE Explorer पैन में, एजेंट की ओर से आपके वर्कस्पेस के रूट में जनरेट की गई दो पाइपलाइन फ़ाइलों की समीक्षा करें:

1. deployment.yaml: इस फ़ाइल को खोलें. यह आपके एनवायरमेंट रजिस्ट्री के तौर पर काम करता है. यह आपके लॉजिकल dev पाइपलाइन को cymbal-airflow एनवायरमेंट से मैप करता है, एक्ज़ीक्यूशन रीजन (us-central1) सेट करता है, और artifact_storage बकेट तय करता है. इस बकेट में कंपाइल किए गए डीएजी और डिपेंडेंसी को स्टेज किया जाता है.
2. fraud_analysis_pipeline.yaml: इस फ़ाइल को खोलें. इससे एक्ज़ीक्यूशन ग्राफ़ तय होता है. इसमें ट्रिगर शेड्यूल (interval: '0 0 * * *') के बारे में बताया गया है. साथ ही, actions ब्लॉक में तीन चरणों को क्रम से दिखाया गया है:
   • Dataproc Serverless पर चल रहे 01_ingestion.ipynb के लिए, डेटा ट्रांसफ़र करने की notebook कार्रवाई.
   • dbt_project डायरेक्ट्री को टारगेट करने वाला ट्रांसफ़ॉर्मेशन pipeline ऐक्शन. इसमें dependsOn डिपेंडेंसी होती है, जो डेटा इंटेक के चरण की ओर इशारा करती है.
   • 03_inference.ipynb के लिए अनुमान लगाने वाली notebook कार्रवाई. इसमें dependsOn डिपेंडेंसी होती है, जो dbt चरण की ओर इशारा करती है. साथ ही, इसमें Spanner JAR प्रॉपर्टी बंडल होती है.
3. इसके अलावा, एजेंट जनरेट किए गए इन आर्टफ़ैक्ट को आपके एडिटर पैनल में मौजूद वॉकट्रू टैब में खास जानकारी के तौर पर दिखाएगा. इसमें किए गए कॉन्फ़िगरेशन और पुष्टि की जानकारी भी शामिल होगी.

इंटरैक्टिव डीएजी कॉन्फ़िगरेशन

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

1. IDE की गतिविधि बार में, Google Cloud Data Agent Kit पैनल खोलें.
2. DATA ENGINEERING में जाकर, Orchestration Pipelines को बड़ा करें.
3. मुख्य एडिटर में विज़ुअल डीएजी कैनवस खोलने के लिए, fraud_analysis_pipeline.yaml पर क्लिक करें.

4. सबसे ऊपर मौजूद Schedule trigger नोड पर क्लिक करें. दाईं ओर कॉन्फ़िगरेशन फ़्लायआउट खुलता है. इसमें पार्स की गई क्रॉन स्ट्रिंग (0 0 * * *) दिखती है. साथ ही, बैकफ़िल और कैचअप जैसे पैरामीटर में बदलाव करने का विकल्प मिलता है.
5. नोटबुक टास्क नोड (जैसे, डेटा ट्रांसफ़र या अनुमान लगाने का चरण) पर क्लिक करें. फ़्लायआउट अपडेट हो जाता है. इसमें Dataproc Serverless के एक्ज़ीक्यूशन की मैपिंग और कनेक्टर की प्रॉपर्टी दिखती हैं.
6. नोड ब्लॉक में, नोटबुक के फ़ाइल नाम वाले हाइपरलिंक (जैसे कि 01_ingestion.ipynb) पर ध्यान दें. इस पर क्लिक करने से, नोटबुक सीधे आपके एडिटर में खुल जाती है.
7. बाईं ओर मौजूद साइडबार में, ऑर्केस्ट्रेशन पाइपलाइन के नीचे, Deployment configuration पर क्लिक करें. इस व्यू में, आपके टारगेट dev एनवायरमेंट क्लस्टर और आउटपुट GCS बकेट के आर्टफ़ैक्ट दिखाए जाते हैं.

सेक्शन की खास जानकारी: आपने एजेंट की मदद से, ऑर्केस्ट्रेशन पाइपलाइन का कॉन्फ़िगरेशन जनरेट किया. इसमें इंटरैक्टिव विज़ुअल कैनवस में, डेटा लेने, dbt, और अनुमान लगाने के टास्क के बीच की डिपेंडेंसी तय की गई.

8. डिप्लॉय करना, लागू करना, और मॉनिटर करना

DAG को स्थानीय तौर पर तय करने के बाद, आपको सेटअप के दौरान उपलब्ध कराए गए मैनेज किए गए Airflow एनवायरमेंट से कनेक्ट करना होगा. इसके बाद, पाइपलाइन को डिप्लॉय करना होगा.

Managed Service for Apache Airflow को कॉन्फ़िगर करना

डेटा एजेंट किट की सेटिंग में, शेड्यूलर कनेक्शन को कॉन्फ़िगर करें. इससे एक्सटेंशन, आपके मैनेज किए जा रहे Airflow एनवायरमेंट को टारगेट करेगा.

1. IDE की गतिविधि बार में, Google Cloud Data Agent Kit पैनल खोलें.
2. SETTINGS में जाकर, सेटिंग पर क्लिक करें.
3. बाईं ओर मौजूद मेन्यू से, शेड्यूलर को चुनें.
4. सेटिंग कॉन्फ़िगर करें:
   • प्रोजेक्ट आईडी: अपना चालू प्रोजेक्ट आईडी चुनें.
   • इलाका: us-central1 को चुनें.
   • एनवायरमेंट: cymbal-airflow को चुनें.
5. सेव करें पर क्लिक करें.

DAG डिप्लॉय करना

डिप्लॉयमेंट आर्टफ़ैक्ट ऑप्टिमाइज़ करना: डिप्लॉय करने से पहले, पक्का करें कि .venv, target/ या logs/ जैसे अस्थायी फ़ोल्डर, आपकी dbt_project/ डायरेक्ट्री में नेस्ट न किए गए हों. हज़ारों फ़ाइलों वाले बड़े वर्चुअल एनवायरमेंट अपलोड करने से, पैकेजिंग में समस्या आ सकती है. साथ ही, Airflow DAG पार्सिंग की प्रोसेस धीमी हो सकती है. अगर ये मौजूद हैं, तो इन्हें हटाएं (rm -rf dbt_project/.venv dbt_project/target dbt_project/logs) या इन्हें .airflowignore फ़ाइल में जोड़ें.

अब कॉन्फ़िगर की गई पाइपलाइन को सीधे विज़ुअल कैनवस से, मैनेज किए जा रहे Airflow एनवायरमेंट में डिप्लॉय करें:

1. Google Cloud Data Agent Kit साइडबार में, DATA ENGINEERING > Orchestration Pipelines को बड़ा करें और विज़ुअल डीएजी कैनवस खोलने के लिए, fraud_analysis_pipeline.yaml पर क्लिक करें.
2. कैनवस टूलबार के सबसे ऊपर दाएं कोने में, नीले रंग के पाइपलाइन चलाएं बटन पर क्लिक करें.
3. एनवायरमेंट के ड्रॉपडाउन पिकर में, dev को चुनें.
4. नीचे मौजूद स्टेटस एरिया (Running pipeline: Building pipeline locally...) में, प्रोसेस की प्रोग्रेस की सूचना देखें. एक्सटेंशन, आपके डीएजी को अपने-आप कंपाइल करेगा. साथ ही, नोटबुक और डीबीटी ऐसेट को पैकेज करेगा और उन्हें आपके मैनेज किए गए Airflow एनवायरमेंट के GCS बकेट में अपलोड करेगा. इसमें करीब तीन से चार मिनट लगते हैं.

रन को मॉनिटर करना

स्थानीय कंपाइलेशन पूरा होने के बाद, पॉप-अप सूचना में Triggered a new run for pipeline... successfully की पुष्टि हो जाती है. इसके बाद, लाइव एक्ज़ीक्यूशन को मॉनिटर करें:

1. Google Cloud Data Agent Kit साइडबार में, DATA ENGINEERING > Orchestration Pipelines को बड़ा करें.
2. पाइपलाइन मैनेजमेंट पर क्लिक करें.
3. पाइपलाइन मैनेजमेंट टेबल में, fraud_analysis_pipeline पर क्लिक करके, पाइपलाइन के एक्ज़ीक्यूशन का इतिहास खोलें.

4. Execution History व्यू में, कैलेंडर से चालू रन चुनें.
5. हर पाइपलाइन टास्क (डेटा इंटेक, डीबीटी ट्रांसफ़ॉर्मेशन, और अनुमान) के पूरा होने पर, स्टेटस इंडिकेटर अपडेट होते हैं और टास्क की अवधि दिखती है. किसी टास्क के लाइव एक्ज़ीक्यूशन का आउटपुट और Airflow DAG लॉग देखने के लिए, उस पर क्लिक करें.

सीआई/सीडी और GitHub Actions का इस्तेमाल करके प्रोडक्शन डिप्लॉयमेंट: लोकल डिप्लॉयमेंट के अलावा, Data Agent Kit अपने-आप GitHub Action वर्कफ़्लो (.github/workflows/deploy.yaml और .github/workflows/validate.yaml) जनरेट करता है. इन्हें GitLab CI या Cloud Build के हिसाब से आसानी से बदला जा सकता है. प्रोडक्शन वर्कफ़्लो के लिए, अपने कोड को कमिट करने और GitHub रिपॉज़िटरी की main ब्रांच में बदलावों को पुश या मर्ज करने से, GitHub ऐक्शन अपने-आप ट्रिगर हो जाता है. इससे पाइपलाइन की पुष्टि की जाती है और उसे प्रोडक्शन Managed Airflow एनवायरमेंट में डिप्लॉय किया जाता है.

सेक्शन की खास जानकारी: आपने Airflow Scheduler कनेक्शन को कॉन्फ़िगर किया, मैनेज किए गए Airflow पर एंड-टू-एंड ऐनलिटिकल पाइपलाइन को डिप्लॉय किया, और लाइव एक्ज़ीक्यूशन की निगरानी की. साथ ही, सिस्टम की पुष्टि की. इसमें रॉ लॉग से लेकर Cloud Spanner की फ़ाइनल अनुमानों तक की जानकारी शामिल है.

9. व्यवस्थित करें

इस कोडलैब में इस्तेमाल किए गए संसाधनों के लिए, Google Cloud प्रोजेक्ट पर लगने वाले शुल्क से बचने के लिए, अपने-आप काम करने वाली स्क्रिप्ट का इस्तेमाल करके एनवायरमेंट को बंद करें.

टियरडाउन को न छोड़ें! Managed Airflow और Cloud Spanner को चालू रखने पर, आपसे शुल्क लिया जाता है. अगर आपने सेटअप को बंद करने की प्रोसेस को छोड़ दिया है, तो इसे बाद में चलाएं या इस लैब के लिए बनाए गए पूरे Google Cloud प्रोजेक्ट को मिटाएं.

1. टर्मिनल पैनल (या Cloud Shell) में, स्क्रिप्ट डायरेक्ट्री पर जाएं और यह कमांड चलाएं:

cd ~/devrel-demos/codelabs/agentic-data-labs/data-science/scripts
chmod +x teardown.sh
./teardown.sh

2. स्क्रिप्ट उन सभी संसाधनों की सूची दिखाएगी जिन्हें मिटाया जाना है. साथ ही, पुष्टि करने के लिए कहेगी:
   • मैनेज किया गया Airflow एनवायरमेंट (cymbal-airflow)
   • Cloud Spanner इंस्टेंस (cymbal-fraud)
   • BigQuery डेटासेट (transactions_dataset_evals)
   • Cloud Storage बकेट (gs://${PROJECT_ID}-fin-clearing-raw और gs://${PROJECT_ID}-models)
   • वर्कर सेवा खाता (composer-worker-sa)
3. पुष्टि करने के लिए, y टाइप करें. टियरडाउन स्क्रिप्ट, GCP की सभी चालू की गई सेवाओं को हटा देगी और लोकल फ़ाइलों को मिटा देगी.

10. बधाई हो!

आपने Cloud Storage, BigQuery, Managed Service for Apache Spark (Spark Serverless), dbt, Cloud Spanner, और Managed Service for Apache Airflow को शामिल करके, धोखाधड़ी का पता लगाने वाली एंड-टू-एंड पाइपलाइन बनाई है. साथ ही, Antigravity IDE में Google Cloud Data Agent Kit के साथ पेयर-प्रोग्रामिंग की है.

आपने क्या-क्या हासिल किया

1. 📥 Managed Service for Apache Spark और Data Agent Kit का इस्तेमाल करके, BigQuery टेबल में इंजस्ट किए गए रॉ लेन-देन के लॉग.
2. 🧹 डेटा क्वालिटी टेस्ट के साथ dbt प्रोजेक्ट बनाकर, डुप्लीकेट डेटा हटाया गया और डेटा को सामान्य बनाया गया.
3. 🤖 डिस्ट्रिब्यूटेड रैंडम फ़ॉरेस्ट मॉडल को ट्रेनिंग दी. इसके लिए, RandomForestClassifier का इस्तेमाल किया गया. साथ ही, ट्रेन किए गए मॉडल को Cloud Storage में एक्सपोर्ट किया.
4. ⚡ आने वाले लेन-देन पर बैच इन्फ़रेंस लागू किया और ज़्यादा जोखिम वाले रिकॉर्ड को ऑडिट की समीक्षा के लिए Cloud Spanner में भेजा.
5. 🔄 Managed Service for Apache Airflow और IDE के विज़ुअल DAG मैनेजमेंट टूल का इस्तेमाल करके, शेड्यूल किए गए Airflow DAG के तौर पर वर्कफ़्लो को व्यवस्थित, डिप्लॉय, और मॉनिटर किया गया.

मुख्य सिद्धांत

सिद्धांत	आपने क्या सीखा
डेटा एजेंट किट	नैचुरल लैंग्वेज का इस्तेमाल करके, IDE में पेयर-प्रोग्रामिंग की सुविधा. इससे PySpark नोटबुक जनरेट की जा सकती हैं, dbt मॉडल कॉन्फ़िगर किए जा सकते हैं, और Airflow DAG तय किए जा सकते हैं
BigQuery	विश्लेषणात्मक एसक्यूएल, डीबीटी ट्रांसफ़ॉर्मेशन, और एमएल ट्रेनिंग के लिए, टेबल के तौर पर डेटा को बड़े पैमाने पर स्टोर करने की सुविधा
Spark Serverless	डेटा को डिस्ट्रिब्यूट करने के लिए PySpark का इस्तेमाल करके, सर्वर के बिना डेटा लोड करना और रैंडम फ़ॉरेस्ट एमएल ट्रेनिंग करना
Cloud Spanner कनेक्टर	स्पार्क इन्फ़रेंस की बैच प्रोसेस से मिले अनुमानों को सीधे तौर पर ऑपरेशनल डेटाबेस की समीक्षा कतारों में लिखना
YAML DAG Declarations	आईडीई में, इंटरैक्टिव Airflow विज़ुअल ग्राफ़ के तौर पर रेंडर की गई डिक्लेरेटिव पाइपलाइन की परिभाषाएं
विज़ुअल डीएजी मैनेजमेंट	पाइपलाइन की डिपेंडेंसी की जांच करना, Managed Airflow पर डिप्लॉय करना, और IDE में लाइव टास्क के एक्ज़ीक्यूशन के इतिहास की निगरानी करना

अगले चरण

• Google Cloud Data Agent Kit के दस्तावेज़ पढ़ें
• Managed Service for Apache Spark के बारे में ज़्यादा जानें
• Managed Service for Apache Airflow के बारे में ज़्यादा जानें
• Antigravity IDE का इस्तेमाल करके, एक से ज़्यादा सेवाओं वाली पाइपलाइन बनाएं