पाइपलाइन रन का शेड्यूल और लॉगिंग
कमांड लाइन से अपनी पाइपलाइन Python स्क्रिप्ट के रूप में चलाएँ, आरंभ और समाप्ति समय लॉग करें और स्वचालन के लिए cron या शेड्यूलर उपयोग करें।
पाइपलाइन रन का शेड्यूल और लॉगिंग, CoddyKit पर Pandas & NumPy Academy का एक निःशुल्क पाठ है। यह 4 में से 4वाँ पाठ है। आप नीचे पूरा पाठ निःशुल्क पढ़ सकते हैं—फिर अंतर्निहित कोड संपादक और 24/7 एआई ट्यूटर के साथ ब्राउज़र में इसका व्यावहारिक अभ्यास कर सकते हैं। यह Pandas & NumPy Academy सीखने के मार्ग का हिस्सा है और आपकी प्रगति वेब तथा CoddyKit ऐप पर सिंक होती रहती है। Pandas & NumPy Academy पाठ्यक्रम में कुल 4 पाठ शामिल हैं।
नोटबुक से स्क्रिप्ट तक
जो पाइपलाइन केवल तब चलती है जब डेवलपर मैन्युअल रूप से कोई नोटबुक खोलता है, वह पहली रन के बाद कोई व्यावसायिक मूल्य नहीं देती। हर दिन स्वचालित रूप से चलाने के लिए पाइपलाइन को ऐसी पाइथन स्क्रिप्ट के रूप में व्यवस्थित करना होगा, जिसे कमांड लाइन से चलाया जा सके: python pipeline.py। इसके लिए if __name__ == '__main__': प्रवेश-बिंदु, कमांड-लाइन तर्कों का पार्सिंग और उचित लॉगिंग आवश्यक हैं—ये उत्पादन स्क्रिप्ट के तीन आधार हैं।
# pipeline.py
import argparse
import logging
import pandas as pd
def main(config_path):
logging.info(f'Starting pipeline with config: {config_path}')
# ... run ETL steps ...
logging.info('Pipeline complete.')
if __name__ == '__main__':
parser = argparse.ArgumentParser()
parser.add_argument('--config', default='config.json')
args = parser.parse_args()
main(args.config)पाइथन लॉगिंग कॉन्फ़िगर करना
पाइपलाइन लॉग के लिए पाइथन का अंतर्निहित logging मॉड्यूल सही साधन है—print() statements नहीं। logging.basicConfig() का उपयोग करके कंसोल आउटपुट और फ़ाइल आउटपुट, दोनों के साथ लॉगर कॉन्फ़िगर करें। सामान्य प्रगति के लिए INFO स्तर और विफलताओं के लिए ERROR स्तर पर लॉग लिखें। फ़ाइल-आधारित लॉग प्रक्रिया समाप्त होने के बाद भी सुरक्षित रहते हैं, जो उन निर्धारित रन की जाँच के लिए आवश्यक है जिन पर कोई निगरानी नहीं कर रहा था।
import logging
from datetime import date
log_file = f'pipeline_{date.today()}.log'
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s %(levelname)s %(message)s',
handlers=[
logging.FileHandler(log_file),
logging.StreamHandler()
]
)
logging.info('Logger configured.')पाइपलाइन के आरंभ और अंत का लॉग रखना
पाइपलाइन रन का आरंभ समय, समाप्ति समय और बीता हुआ समय हमेशा लॉग करें। इससे एक आधाररेखा बनती है: यदि पाइपलाइन सामान्यतः 45 सेकंड में चलती है और आज 8 मिनट लगे, तो कुछ बदला है—शायद इनपुट फ़ाइल 10 गुना बड़ी है या डेटाबेस query धीमी चल रही है। टाइमस्टैम्प वाले आरंभ और समाप्ति लॉग प्रविष्टियों से केवल लॉग फ़ाइल देखकर यह तुलना आसानी से की जा सकती है।
import time
import logging
def run_pipeline(config):
start = time.time()
logging.info(f'Pipeline START | env={config.get("env", "dev")} | input={config["input_path"]}')
try:
df = extract(config)
df_clean = transform(df, config)
load(df_clean, config)
elapsed = time.time() - start
logging.info(f'Pipeline SUCCESS | rows={len(df_clean)} | elapsed={elapsed:.1f}s')
except Exception as e:
logging.error(f'Pipeline FAILED | error={e}', exc_info=True)
raiseहर चरण की पंक्ति-संख्या का लॉग रखना
हर रूपांतरण चरण में प्रवेश करने और उससे निकलने वाली पंक्तियों की संख्या का लॉग रखें। एक साफ़ लॉग ऐसा दिखता है: extract: 50,000 rows → drop_nulls: 49,200 rows → filter: 47,800 rows → output: 47,800 rows। इस रिकॉर्ड से तुरंत स्पष्ट हो जाता है कि हर चरण में कितनी पंक्तियाँ हटाई गईं और संख्याएँ अपेक्षित हैं या नहीं। असामान्य कमी लॉग की गई संख्याओं के बीच अंतर के रूप में दिखाई देती है।
def log_step(df, step_name):
logging.info(f'{step_name}: {len(df):,} rows')
return df
import pandas as pd
df = (pd.read_csv('orders.csv')
.pipe(log_step, 'extract')
.dropna(subset=['revenue'])
.pipe(log_step, 'drop_nulls')
.query('quantity > 0')
.pipe(log_step, 'filter_qty')
)
print('Step logging complete.')Linux/Mac पर cron के साथ शेड्यूल करना
cron बार-बार चलने वाले कार्यों के लिए मानक Unix शेड्यूलर है। crontab -e से crontab संपादित करें और स्क्रिप्ट कब चलानी है, यह बताने वाली एक पंक्ति जोड़ें। प्रारूप है: minute hour day month weekday command। 6:00 AM पर हर दिन चलने वाली पाइपलाइन के लिए 0 6 * * * /usr/bin/python /path/to/pipeline.py का उपयोग करें। cron प्रविष्टियों में हमेशा पूर्ण पथों का उपयोग करें, क्योंकि cron आपके शेल की PATH सेटिंग के बिना न्यूनतम वातावरण में चलता है।
# crontab entry — edit with: crontab -e
# Run pipeline.py at 06:00 every day
# 0 6 * * * /opt/homebrew/bin/python /Users/analyst/pipeline.py --config /Users/analyst/config.json >> /Users/analyst/cron.log 2>&1
# Common cron patterns:
# 0 6 * * * — daily at 06:00
# 0 */4 * * * — every 4 hours
# 0 9 * * 1 — every Monday at 09:00
print('Cron schedule format: minute hour day month weekday')पाइथन schedule लाइब्रेरी के साथ शेड्यूल करना
schedule लाइब्रेरी cron को छुए बिना निर्धारित अंतरालों पर कार्य चलाने का शुद्ध-पाइथन तरीका उपलब्ध कराती है। यह उन वातावरणों में उपयोगी है जहाँ cron उपलब्ध नहीं है (Windows), या जब आप शेड्यूलर का तर्क स्वयं पाइथन प्रक्रिया के अंदर रखना चाहते हैं। पाइपलाइन को निर्धारित कार्य-लूप में रखें और प्रक्रिया को चालू रखें, ताकि वह बार-बार निष्पादित हो सके।
# pip install schedule
# import schedule, time
# def job():
# logging.info('Scheduled run starting...')
# run_pipeline(CONFIG)
# schedule.every().day.at('06:00').do(job)
# schedule.every(4).hours.do(job)
# while True:
# schedule.run_pending()
# time.sleep(60)
print('schedule library: use for in-process Python scheduling')त्रुटि प्रबंधन और निकास कोड
पाइपलाइन स्क्रिप्ट विफल होने पर शून्य से अलग निकास कोड लौटाए, ताकि शेड्यूलर को पता चले कि कार्य विफल हुआ है। मुख्य निष्पादन को try/except ब्लॉक में रखें और विफलता पर sys.exit(1) कॉल करें। cron, Jenkins और Airflow सभी निकास कोड जाँचते हैं: शून्य से अलग कोड चेतावनी, दोबारा रन या सूचना को सक्रिय करता है। ऐसी अनियंत्रित exception, जो निकास कोड निर्धारित नहीं करती, स्वचालित निगरानी से अनदेखी रह सकती है।
import sys
def main():
try:
run_pipeline(CONFIG)
sys.exit(0) # success
except AssertionError as e:
logging.error(f'Data validation failed: {e}')
sys.exit(2) # data error
except Exception as e:
logging.error(f'Unexpected error: {e}', exc_info=True)
sys.exit(1) # general failure
print('Exit code 0=success, 1=error, 2=data failure')पाइपलाइन रन की सारांश फ़ाइल लिखना
सफल रन के बाद आउटपुट के साथ एक छोटी JSON सारांश फ़ाइल लिखें। इसमें रन का टाइमस्टैम्प, इनपुट पंक्तियों की संख्या, आउटपुट पंक्तियों की संख्या, हटाई गई पंक्तियाँ और बीता हुआ समय शामिल करें। निगरानी प्रणालियाँ और डैशबोर्ड समय के साथ पाइपलाइन के स्वास्थ्य के रुझानों पर नज़र रखने के लिए इस फ़ाइल को पढ़ सकते हैं। पिछले 30 दिनों में आउटपुट पंक्तियाँ दिखाने वाला डैशबोर्ड यह पता लगाना आसान बनाता है कि डेटा स्रोत ने किस दिन कम रिकॉर्ड भेजना शुरू किया।
import json
from datetime import datetime
def write_run_summary(config, input_rows, output_rows, elapsed):
summary = {
'run_at': datetime.now().isoformat(),
'input_path': config['input_path'],
'input_rows': input_rows,
'output_rows': output_rows,
'rows_dropped': input_rows - output_rows,
'elapsed_seconds': round(elapsed, 2),
'status': 'success'
}
with open('last_run_summary.json', 'w') as f:
json.dump(summary, f, indent=2)
print('Run summary written.')Idempotent शेड्यूलिंग: दोहरी रन से बचना
यदि निर्धारित पाइपलाइन गलती से दो बार शुरू हो जाए, तो उसे आउटपुट खराब नहीं करना चाहिए। लोड चरण को idempotent बनाएँ: तारीख वाला आउटपुट फ़ाइल नाम उपयोग करें या नवीनतम परिणाम से उसी आउटपुट को अधिलेखित करें। डेटाबेस लोड के लिए if_exists='replace' या UPSERT पैटर्न का उपयोग करें। डुप्लिकेट हटाने वाले चरण के बिना कभी भी append मोड का उपयोग न करें, वरना हर निर्धारित रन आउटपुट तालिका में डुप्लिकेट पंक्तियाँ जोड़ देगा।
from datetime import date
def load_idempotent(df, config):
# Date-stamped output: each run overwrites its own day's file
output_path = f"output_{date.today().strftime('%Y%m%d')}.parquet"
df.to_parquet(output_path, index=False)
logging.info(f'Loaded {len(df)} rows to {output_path}')पाइपलाइन विफल होने पर अलर्ट
जिन पाइपलाइनों पर व्यावसायिक संचालन निर्भर करते हैं, उनमें विफलता के बाद कोई सूचना न मिलना खतरनाक है। एक सरल अलर्ट सेट करें: यदि अपेक्षित समय-सीमा के भीतर रन सारांश फ़ाइल अपडेट न हो, तो ईमेल या Slack संदेश भेजें। विफलता होने पर ईमेल भेजने के लिए पाइथन का smtplib इस्तेमाल किया जा सकता है, या Slack पर पोस्ट करने के लिए वेबहुक का उपयोग किया जा सकता है। निकास कोड 1 या 2 मिलते ही अलर्ट भेजें, ताकि विश्लेषक को पता चल जाए कि दैनिक रिफ्रेश छूट गया है, इससे पहले कि व्यवसाय को इसका पता चले।
import smtplib
def send_failure_alert(error_msg):
# Example: send plain-text email via SMTP
# server = smtplib.SMTP('smtp.example.com', 587)
# server.sendmail('pipeline@company.com',
# 'analyst@company.com',
# f'Subject: Pipeline Failed\n\n{error_msg}')
# server.quit()
print(f'[ALERT] Would send failure notification: {error_msg}')
# In main():
# except Exception as e:
# send_failure_alert(str(e))
# sys.exit(1)
print('Alert integration pattern shown above.')निर्धारित पाइपलाइन स्क्रिप्ट पूरी करना
सभी हिस्सों—आर्ग्युमेंट पार्सिंग, लॉगिंग कॉन्फ़िगरेशन, रन सारांश, त्रुटि प्रबंधन और निकास कोड—को मिलाकर एक पूरी पाइपलाइन स्क्रिप्ट बनाएं। इस स्क्रिप्ट को किसी भी वातावरण में रखा जा सकता है, कॉन्फ़िगरेशन फ़ाइल से जोड़ा जा सकता है और क्रॉन या किसी वर्कफ़्लो ऑर्केस्ट्रेटर के साथ निर्धारित किया जा सकता है। हर निष्पादन पर यह तारीख वाली लॉग फ़ाइल, रन सारांश और तारीख वाली आउटपुट फ़ाइल बनाती है, जिससे हर रन का पूरा ऑडिट और स्वतंत्र रूप से पुनरुत्पादन संभव होता है।
# Full script skeleton:
# 1. parse --config argument
# 2. configure logging to file + console
# 3. load JSON config
# 4. validate config
# 5. run extract() -> transform() -> load()
# 6. write run summary JSON
# 7. sys.exit(0) on success, sys.exit(1) on failure
print('Production pipeline script structure complete.')
print('Schedule with: crontab -e or python scheduler.py')त्वरित जाँच
इस पाठ से डेटा विश्लेषण की अवधारणाओं की अपनी समझ जाँचें।
पाठ का पुनरावलोकन
इस पाठ में आपने सीखा: आर्ग्युमेंट पार्सिंग और लॉगिंग के साथ पाइपलाइन को कमांड-लाइन स्क्रिप्ट के रूप में व्यवस्थित करना, क्रॉन के साथ शेड्यूल करना और गैर-शून्य निकास कोड तथा अलर्ट के माध्यम से विफलताओं को संभालना, और विश्वसनीय स्वचालित निष्पादन के लिए रन सारांश फ़ाइलें लिखना तथा आइडेम्पोटेंट लोड चरणों की रूपरेखा बनाना। डेटा विश्लेषण: Pandas और NumPy ट्रैक पूरा करने पर बधाई!
एआई शिक्षक के साथ Python सीखें — निःशुल्क
अपने ब्राउज़र में वास्तविक कोड लिखें और चलाएँ, चौबीसों घंटे एआई शिक्षक से तुरंत सहायता पाएँ, और वेब या ऐप पर वहीं से शुरू करें जहाँ आपने छोड़ा था।
- पाठ्यक्रम
- 30
- पाठ
- 120
अक्सर पूछे जाने वाले प्रश्न
क्या “पाइपलाइन रन का शेड्यूल और लॉगिंग” पाठ निःशुल्क है?
हाँ—“पाइपलाइन रन का शेड्यूल और लॉगिंग” का पूरा पाठ यहाँ वेब पर निःशुल्क पढ़ा जा सकता है। इंटरैक्टिव अभ्यास (अंतर्निहित कोड संपादक और 24/7 एआई ट्यूटर) करने और Pandas & NumPy Academy पाठ्यक्रम का बाकी हिस्सा अनलॉक करने के लिए CoddyKit PRO लें। Pandas & NumPy Academy पाठ्यक्रम में कुल 4 पाठ शामिल हैं।
“पाइपलाइन रन का शेड्यूल और लॉगिंग” में मैं क्या सीखूँगा?
कमांड लाइन से अपनी पाइपलाइन Python स्क्रिप्ट के रूप में चलाएँ, आरंभ और समाप्ति समय लॉग करें और स्वचालन के लिए cron या शेड्यूलर उपयोग करें। आप ब्राउज़र में सीधे चलाए जाने वाले व्यावहारिक कोड के साथ Pandas & NumPy Academy का अभ्यास करते हैं, और पाठ पूरा करते समय 24/7 एआई ट्यूटर आपके प्रश्नों के उत्तर देता है।
क्या Pandas & NumPy Academy शुरू करने के लिए मुझे किसी अनुभव की आवश्यकता है?
पहले के अनुभव की आवश्यकता नहीं है। CoddyKit पर Pandas & NumPy Academy शुरुआती से लेकर उन्नत शिक्षार्थियों तक सभी के लिए व्यवस्थित किया गया है, इसलिए आप यहीं से या शुरुआत से सीखना शुरू कर सकते हैं और अपनी गति से आगे बढ़ सकते हैं। यह 4 में से 4वाँ पाठ है।
“पाइपलाइन रन का शेड्यूल और लॉगिंग” पाठ पूरा करने में कितना समय लगता है?
CoddyKit का अधिकांश पाठ लगभग 5–10 मिनट में पूरा हो जाता है। हर पाठ छोटा और संवादात्मक है, इसलिए आप लगातार प्रगति करते हैं और वेब या ऐप पर वहीं से सीखना जारी रख सकते हैं जहाँ आपने छोड़ा था।
क्या मैं इस Pandas & NumPy Academy पाठ में कोड लिख और चला सकता हूँ?
हाँ। हर Pandas & NumPy Academy पाठ में एक अंतर्निर्मित कोड संपादक शामिल है, जिससे आप सीधे अपने ब्राउज़र में वास्तविक कोड लिख और चला सकते हैं और तुरंत एआई प्रतिक्रिया पा सकते हैं—स्थानीय सेटअप की आवश्यकता नहीं है।
इस पाठ्यक्रम के सभी पाठ
- रूपांतरण चरणों को फ़ंक्शन के रूप में व्यवस्थित करना
- Config dict से शृंखलाओं में पैरामीटर जोड़ना
- अभिकथनों से पाइपलाइन चरणों का परीक्षण
- पाइपलाइन रन का शेड्यूल और लॉगिंग