İşlem Hattı Çalıştırmalarını Zamanlama ve Günlüğe Kaydetme
İşlem hattınızı komut satırından bir Python betiği olarak çalıştırın, başlangıç ve bitiş zamanlarını günlüğe kaydedin ve otomasyon için cron veya bir zamanlayıcı kullanın.
İşlem Hattı Çalıştırmalarını Zamanlama ve Günlüğe Kaydetme, CoddyKit'te ücretsiz bir Pandas & NumPy Academy dersidir. Bu, 4 dersinin 4. dersidir. Aşağıdan dersin tamamını ücretsiz okuyabilir, sonra tarayıcıda yerleşik kod editörü ve 7/24 yapay zeka koçu ile uygulamalı olarak pratik yapabilirsin. Bu, Pandas & NumPy Academy öğrenme yolunun bir parçasıdır ve ilerlemeniz web ve CoddyKit uygulaması arasında senkronize olur. Pandas & NumPy Academy kursu toplamda 4 dersten oluşur.
Not Defterinden Betiğe
Yalnızca bir geliştirici not defterini elle açtığında çalışan bir pipeline, ilk çalıştırmanın ötesinde hiçbir iş değeri sağlamaz. Pipeline'ın her gün otomatik çalışması için komut satırından yürütülebilen bir Python betiği olarak yapılandırılması gerekir: python pipeline.py. Bunun için if __name__ == '__main__': giriş noktası, komut satırı bağımsız değişkenlerini ayrıştırma ve uygun günlük kaydı gerekir; bunlar üretim betiğinin üç temel unsurudur.
# 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)Python Günlük Kaydını Yapılandırma
Python'un yerleşik logging modülü, pipeline günlükleri için doğru araçtır; print() ifadeleri değil. logging.basicConfig() kullanarak hem konsol hem de dosya çıktısına sahip bir günlük kaydedici yapılandırın. Normal ilerleme için INFO, hatalar için ERROR düzeyinde günlük kaydı yapın. Dosya tabanlı günlükler işlem sona erdikten sonra da korunur; bu, kimsenin izlemediği zamanlanmış çalıştırmalarda hata ayıklamak için zorunludur.
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.')Pipeline Başlangıç ve Bitişini Günlüğe Kaydetme
Bir pipeline çalıştırmasının başlangıç zamanını, bitiş zamanını ve geçen süresini her zaman günlüğe kaydedin. Bu, bir karşılaştırma temeli oluşturur: Pipeline normalde 45 saniyede çalışıyor ve bugün 8 dakika sürdüyse bir şey değişmiştir; örneğin girdi dosyası 10 kat büyümüş veya bir veritabanı sorgusu yavaş çalışıyor olabilir. Zaman damgalı başlangıç ve bitiş günlük kayıtları, bu karşılaştırmayı yalnızca günlük dosyasından yapmayı kolaylaştırır.
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)
raiseHer Adım İçin Satır Sayılarını Günlüğe Kaydetme
Her dönüştürme adımına giren ve çıkan satır sayısını günlüğe kaydedin. Düzenli bir günlük şöyle görünür: extract: 50,000 satır → drop_nulls: 49,200 satır → filter: 47,800 satır → output: 47,800 satır. Bu iz, her adımda kaç satırın elendiğini ve sayıların beklenen düzeyde olup olmadığını hemen anlaşılır kılar. Olağandışı düşüşler, günlüğe kaydedilen sayılardaki boşluklar olarak ortaya çıkar.
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'te cron ile Zamanlama
cron, tekrarlanan işleri zamanlamak için standart Unix zamanlayıcısıdır. crontab -e ile crontab'ı düzenleyin ve betiğin ne zaman çalıştırılacağını belirten bir satır ekleyin. Biçim şöyledir: dakika saat gün ay haftanın günü komut. Her gün saat 6:00'da çalışması gereken bir pipeline için 0 6 * * * /usr/bin/python /path/to/pipeline.py kullanılır. cron girişlerinde her zaman mutlak yollar kullanın; çünkü cron, kabuğunuzun PATH ayarlarının bulunmadığı sınırlı bir ortamda çalışır.
# 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')Python schedule Kütüphanesiyle Zamanlama
schedule kütüphanesi, cron'a dokunmadan işleri belirli aralıklarla çalıştırmak için yalnızca Python kullanan bir yöntem sağlar. cron'un kullanılamadığı ortamlarda (Windows) veya zamanlayıcı mantığını Python sürecinin kendisinde tutmak istediğinizde kullanışlıdır. Pipeline'ı zamanlanmış bir iş döngüsüne sarın ve tekrar tekrar çalışabilmesi için süreci canlı tutun.
# 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')Hata İşleme ve Çıkış Kodları
Bir pipeline betiği başarısız olduğunda zamanlayıcının işin başarısız olduğunu anlaması için sıfır olmayan bir çıkış kodu döndürmelidir. Ana yürütmeyi bir try/except bloğuna sarın ve başarısızlık durumunda sys.exit(1) çağırın. cron, Jenkins ve Airflow çıkış kodunu denetler: sıfır olmayan bir kod uyarı, yeniden çalıştırma veya bildirim tetikler. Çıkış kodunu ayarlamayan, işlenmemiş bir istisna otomatik izleme tarafından fark edilmeyebilir.
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')Pipeline Çalıştırma Özeti Dosyası Yazma
Başarılı bir çalıştırmadan sonra çıktının yanına küçük bir JSON özet dosyası yazın. Çalıştırma zaman damgasını, girdi satırı sayısını, çıktı satırı sayısını, silinen satırları ve geçen süreyi ekleyin. İzleme sistemleri ve panolar, pipeline'ın zaman içindeki durum eğilimlerini takip etmek için bu dosyayı okuyabilir. Son 30 gündeki çıktı satırlarını gösteren bir pano, veri kaynağının daha az kayıt göndermeye başladığı günü kolayca fark ettirir.
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.')Yinelemeli Zamanlama: Çift Çalıştırmaları Önleme
Zamanlanmış bir pipeline yanlışlıkla iki kez tetiklenirse çıktıyı bozmamalıdır. Yükleme adımını yinelemeli olacak şekilde tasarlayın: tarih içeren bir çıktı dosyası adı kullanın veya aynı çıktının üzerine en son sonucu yazın. Veritabanı yüklemelerinde if_exists='replace' ya da UPSERT düzeni kullanın. Tekilleştirme adımı olmadan hiçbir zaman append kipini kullanmayın; aksi halde her zamanlanmış çalıştırma çıktı tablosuna yinelenen satırlar ekler.
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}')Pipeline Başarısızlıklarında Uyarı Gönderme
İş operasyonlarının bağlı olduğu pipeline'larda başarısızlık sonrası sessizlik tehlikelidir. Basit bir uyarı düzeni kurun: çalıştırma özeti dosyası beklenen zaman aralığında güncellenmezse e-posta veya Slack mesajı gönderin. Python'un smtplib modülü başarısızlık durumunda e-posta gönderebilir ya da Slack'e gönderi yapmak için bir web kancası kullanabilirsiniz. Analistin günlük yenilemenin başarısız olduğunu işletme fark etmeden öğrenmesi için çıkış kodu 1 veya 2 olduğunda hemen uyarı gönderin.
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.')Eksiksiz Zamanlanmış Pipeline Betiği
Tüm parçaları — bağımsız değişken ayrıştırma, günlük kaydı yapılandırması, çalıştırma özeti, hata işleme ve çıkış kodlarını — eksiksiz bir pipeline betiğinde birleştirin. Bu betik herhangi bir ortama aktarılabilir, bir yapılandırma dosyasına yönlendirilebilir ve cron ya da herhangi bir iş akışı düzenleyicisiyle zamanlanabilir. Her çalıştırmada tarih içeren bir günlük dosyası, çalıştırma özeti ve tarih içeren bir çıktı dosyası üretir; böylece her çalıştırma tamamen denetlenebilir ve bağımsız olarak yeniden üretilebilir.
# 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')Hızlı Kontrol
Bu derste ele alınan Veri Analizi kavramlarını anlayıp anlamadığınızı test edin.
Ders Özeti
Bu derste şunları öğrendiniz: bir pipeline'ı bağımsız değişken ayrıştırma ve günlük kaydı içeren bir komut satırı betiği olarak yapılandırmayı, cron ile zamanlamayı ve sıfır olmayan çıkış kodları ile uyarıları kullanarak başarısızlıkları ele almayı ve güvenilir otomatik yürütme için çalıştırma özeti dosyaları yazmayı ve yinelemeli yükleme adımları tasarlamayı. Veri Analizi: Pandas ve NumPy yolunu tamamladığınız için tebrikler!
Yapay zeka eğitmeniyle Python öğren — ücretsiz
Tarayıcında gerçek kod yaz ve çalıştır, 7/24 yapay zeka eğitmeninden anında yardım al; web'de ya da uygulamada kaldığın yerden devam et.
- Kurslar
- 30
- Dersler
- 120
Sıkça Sorulan Sorular
“İşlem Hattı Çalıştırmalarını Zamanlama ve Günlüğe Kaydetme” dersi ücretsiz mi?
Evet — “İşlem Hattı Çalıştırmalarını Zamanlama ve Günlüğe Kaydetme” dersin tüm metni burada web'de ücretsiz olarak okunabilir. Etkileşimli olarak pratik yapmak (yerleşik kod editörü ve 7/24 yapay zeka koçu) ve Pandas & NumPy Academy kursunun geri kalanını açmak için CoddyKit PRO'ya yükselt. Pandas & NumPy Academy kursu toplamda 4 dersten oluşur.
“İşlem Hattı Çalıştırmalarını Zamanlama ve Günlüğe Kaydetme” dersinde ne öğreneceğim?
İşlem hattınızı komut satırından bir Python betiği olarak çalıştırın, başlangıç ve bitiş zamanlarını günlüğe kaydedin ve otomasyon için cron veya bir zamanlayıcı kullanın. Pandas & NumPy Academy ile uygulamalı kodu tarayıcıda doğrudan çalıştırarak pratik yaparsın ve 7/24 yapay zeka koçu dersi çalışırken sorularını yanıtlar.
Pandas & NumPy Academy öğrenmeye başlamak için deneyim gerekli mi?
Önceden deneyim gerekmez. CoddyKit'te Pandas & NumPy Academy, başlangıçtan ileri seviyeye kadar yapılandırıldığı için buradan başlayabilir veya başından başlayıp kendi hızında ilerleme yapabilirsin. Bu, 4 dersinin 4. dersidir.
“İşlem Hattı Çalıştırmalarını Zamanlama ve Günlüğe Kaydetme” dersi ne kadar sürer?
Çoğu CoddyKit dersi yaklaşık 5–10 dakika sürer. Her biri kısa ve etkileşimli olduğu için sabit ilerleme yaparsın ve web ile uygulama arasında tam olarak bıraktığın yerden devam edebilirsin.
Bu Pandas & NumPy Academy dersinde kod yazıp çalıştırabilir miyim?
Evet. Her Pandas & NumPy Academy dersi yerleşik bir kod editörü içerir, bu sayede tarayıcıda gerçek kod yazıp çalıştırabilir ve anlık yapay zeka geri bildirimi alırsın — yerel kurulum gerekli değildir.
Bu kursun tüm dersleri
- Dönüşüm Adımlarını İşlevler Olarak Yapılandırma
- İşlem Hatlarını Yapılandırma Sözlükleriyle Parametreleştirme
- İşlem Hattı Adımlarını Assertion'larla Test Etme
- İşlem Hattı Çalıştırmalarını Zamanlama ve Günlüğe Kaydetme