Pandas & NumPy Academy · Pelajaran

Menjadualkan dan Merekod Pelaksanaan Pipeline

Jalankan pipeline sebagai skrip Python daripada baris perintah, rekodkan masa mula dan tamat, serta gunakan cron atau penjadual untuk automasi.

Pelajaran 4 daripada 413 langkah

Menjadualkan dan Merekod Pelaksanaan Pipeline ialah pelajaran Pandas & NumPy Academy percuma di CoddyKit. Ini ialah pelajaran 4 daripada 4. Anda boleh membaca keseluruhan pelajaran di bawah secara percuma — kemudian berlatih secara praktikal dalam pelayar menggunakan penyunting kod terbina dalam dan tutor kecerdasan buatan 24/7. Pelajaran ini merupakan sebahagian daripada laluan pembelajaran Pandas & NumPy Academy, dan kemajuan anda disegerakkan merentas web serta aplikasi CoddyKit. Kursus Pandas & NumPy Academy merangkumi sejumlah 4 pelajaran.

Daripada Notebook kepada Skrip

Pipeline yang hanya berjalan apabila pembangun membuka notebook secara manual tidak memberikan nilai perniagaan selain daripada pelaksanaan pertama. Untuk berjalan secara automatik setiap hari, pipeline mesti distrukturkan sebagai skrip Python yang boleh dilaksanakan daripada baris perintah: python pipeline.py. Ini memerlukan titik masuk if __name__ == '__main__':, penghuraian argumen baris perintah dan pengelogan yang betul — tiga tunjang skrip produksi.

# 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)

Mengkonfigurasi Pengelogan Python

Modul logging terbina dalam Python ialah alat yang betul untuk log pipeline — bukannya pernyataan print(). Konfigurasikan pengelog dengan output konsol dan output fail menggunakan logging.basicConfig(). Log pada aras INFO untuk kemajuan biasa dan ERROR untuk kegagalan. Log berasaskan fail kekal selepas proses keluar, yang penting untuk menyahpepijat pelaksanaan berjadual yang tidak dipantau oleh sesiapa.

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.')

Mengelog Permulaan dan Penghujung Pipeline

Sentiasa log masa mula, masa tamat dan masa berlalu bagi sesuatu pelaksanaan pipeline. Ini menyediakan garis dasar: jika pipeline biasanya berjalan selama 45 saat tetapi hari ini mengambil masa 8 minit, sesuatu telah berubah — mungkin fail input 10 kali lebih besar atau pertanyaan pangkalan data berjalan perlahan. Entri log mula dan tamat yang mempunyai cap masa menjadikan perbandingan ini mudah hanya daripada fail log.

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

Mengelog Bilangan Baris bagi Setiap Langkah

Log bilangan baris yang masuk dan keluar bagi setiap langkah transformasi. Log yang kemas kelihatan begini: extract: 50,000 baris → drop_nulls: 49,200 baris → filter: 47,800 baris → output: 47,800 baris. Jejak ini menjelaskan serta-merta bilangan baris yang dibuang pada setiap langkah dan sama ada bilangannya dijangka. Pembuangan luar biasa akan kelihatan sebagai jurang dalam bilangan yang dilog.

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.')

Penjadualan dengan cron pada Linux/Mac

cron ialah penjadual Unix standard untuk kerja berulang. Edit crontab dengan crontab -e dan tambah satu baris yang menyatakan masa untuk menjalankan skrip. Formatnya ialah: minit jam hari bulan hari-minggu arahan. Pipeline yang mesti berjalan pada 6:00 AM setiap hari menggunakan 0 6 * * * /usr/bin/python /path/to/pipeline.py. Sentiasa gunakan laluan mutlak dalam entri cron kerana cron berjalan dalam persekitaran minimum tanpa tetapan PATH shell anda.

# 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')

Penjadualan dengan Pustaka schedule Python

Pustaka schedule menyediakan cara berasaskan Python sepenuhnya untuk menjalankan kerja pada selang masa yang ditetapkan tanpa menggunakan cron. Ia berguna dalam persekitaran yang tidak menyediakan cron (Windows) atau apabila anda mahu logik penjadual berada dalam proses Python itu sendiri. Bungkus pipeline dalam gelung kerja berjadual dan kekalkan proses hidup supaya ia dilaksanakan berulang kali.

# 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')

Pengendalian Error dan Kod Keluar

Skrip pipeline hendaklah mengembalikan kod keluar bukan sifar apabila gagal supaya penjadual mengetahui kerja itu gagal. Bungkus pelaksanaan utama dalam blok try/except dan panggil sys.exit(1) apabila gagal. cron, Jenkins dan Airflow semuanya menyemak kod keluar: kod bukan sifar mencetuskan amaran, pelaksanaan semula atau pemberitahuan. Pengecualian yang tidak dikendalikan dan tidak menetapkan kod keluar mungkin tidak disedari oleh pemantauan automatik.

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')

Menulis Fail Ringkasan Pelaksanaan Pipeline

Selepas pelaksanaan yang berjaya, tulis fail ringkasan JSON kecil bersama output. Sertakan cap masa pelaksanaan, bilangan baris input, bilangan baris output, baris yang dibuang dan masa berlalu. Sistem pemantauan dan papan pemuka boleh membaca fail ini untuk menjejaki trend kesihatan pipeline dari semasa ke semasa. Papan pemuka yang memaparkan baris output sepanjang 30 hari lalu memudahkan pengesanan hari apabila sumber data mula menghantar lebih sedikit rekod.

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.')

Penjadualan Idempoten: Mengelakkan Pelaksanaan Berganda

Jika pipeline berjadual tercetus dua kali secara tidak sengaja, ia tidak sepatutnya merosakkan output. Reka langkah pemuatan supaya idempoten: gunakan nama fail output bertarikh atau tulis ganti output yang sama dengan hasil terkini. Untuk pemuatan pangkalan data, gunakan if_exists='replace' atau corak UPSERT. Jangan gunakan mod append tanpa langkah penyahgandaan, kerana setiap pelaksanaan berjadual akan menambah baris pendua ke jadual output.

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}')

Memberi Amaran tentang Kegagalan Pipeline

Bagi pipeline yang menjadi sandaran operasi perniagaan, tiada pemberitahuan selepas kegagalan adalah berbahaya. Sediakan amaran ringkas: jika fail ringkasan pelaksanaan tidak dikemas kini dalam tempoh yang dijangka, hantar e-mel atau mesej Slack. Python's smtplib boleh menghantar e-mel apabila berlaku kegagalan, atau Anda boleh menggunakan webhook untuk menghantar kiriman ke Slack. Berikan amaran serta-merta apabila kod keluar ialah 1 atau 2 supaya penganalisis mengetahui bahawa penyegaran harian terlepas sebelum pihak perniagaan menyedarinya.

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.')

Skrip Pipeline Berjadual Lengkap

Gabungkan semua bahagian — penghuraian argumen, konfigurasi logging, ringkasan pelaksanaan, pengendalian ralat dan kod keluar — menjadi skrip pipeline yang lengkap. Skrip ini boleh dimasukkan ke dalam mana-mana persekitaran, ditetapkan kepada fail konfigurasi dan dijadualkan dengan cron atau mana-mana pengatur tugas aliran kerja. Skrip ini menghasilkan fail log bertarikh, ringkasan pelaksanaan dan fail output bertarikh bagi setiap pelaksanaan, menjadikan setiap pelaksanaan boleh diaudit sepenuhnya dan dihasilkan semula secara berasingan.

# 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')

Semakan Ringkas

Uji pemahaman Anda tentang konsep Analisis Data daripada pelajaran ini.

Jawab dalam pelajaran interaktif

Imbas Kembali Pelajaran

Dalam pelajaran ini, Anda mempelajari: menstrukturkan pipeline sebagai skrip baris perintah dengan penghuraian argumen dan logging, menjadualkan dengan cron serta mengendalikan kegagalan menggunakan kod keluar bukan sifar dan amaran, dan menulis fail ringkasan pelaksanaan serta mereka bentuk langkah pemuatan idempoten untuk pelaksanaan automatik yang boleh dipercayai. Tahniah kerana telah melengkapkan laluan Analisis Data: Pandas dan NumPy!

Percuma untuk bermula

Pelajari Python dengan tutor kecerdasan buatan — percuma

Tulis dan jalankan kod sebenar dalam pelayar anda, dapatkan bantuan segera daripada tutor kecerdasan buatan yang tersedia 24/7, dan sambung semula dari tempat anda berhenti di web atau dalam aplikasi.

Kursus: 30
Pelajaran: 120

Mula belajar secara percuma Lihat kursus penuh

Soalan Lazim

Adakah pelajaran “Menjadualkan dan Merekod Pelaksanaan Pipeline” percuma?

Ya — teks penuh “Menjadualkan dan Merekod Pelaksanaan Pipeline” boleh dibaca secara percuma di web ini. Untuk berlatih secara interaktif menggunakan penyunting kod terbina dalam dan tutor kecerdasan buatan 24/7, serta membuka kunci baki kursus Pandas & NumPy Academy, tingkat taraf kepada CoddyKit PRO. Kursus Pandas & NumPy Academy merangkumi sejumlah 4 pelajaran.

Apakah yang akan saya pelajari dalam “Menjadualkan dan Merekod Pelaksanaan Pipeline”?

Jalankan pipeline sebagai skrip Python daripada baris perintah, rekodkan masa mula dan tamat, serta gunakan cron atau penjadual untuk automasi. Anda berlatih Pandas & NumPy Academy menggunakan kod praktikal yang dijalankan terus dalam pelayar, manakala tutor kecerdasan buatan 24/7 menjawab soalan anda semasa anda mengikuti pelajaran.

Adakah saya memerlukan pengalaman untuk memulakan Pandas & NumPy Academy?

Tiada pengalaman terdahulu diperlukan. Pembelajaran Pandas & NumPy Academy di CoddyKit disusun untuk pelajar daripada peringkat pemula hingga lanjutan, jadi anda boleh bermula di sini atau dari awal dan belajar mengikut kadar anda sendiri. Ini ialah pelajaran 4 daripada 4.

Berapa lamakah pelajaran “Menjadualkan dan Merekod Pelaksanaan Pipeline” diambil?

Kebanyakan pelajaran CoddyKit mengambil masa kira-kira 5–10 minit. Setiap pelajaran ringkas dan interaktif, jadi anda boleh membuat kemajuan secara berterusan dan menyambung tepat dari tempat anda berhenti di web atau aplikasi.

Bolehkah saya menulis dan menjalankan kod dalam pelajaran Pandas & NumPy Academy ini?

Ya. Setiap pelajaran Pandas & NumPy Academy menyertakan penyunting kod terbina dalam, jadi anda boleh menulis dan menjalankan kod sebenar terus dalam pelayar serta menerima maklum balas kecerdasan buatan serta-merta — tanpa memerlukan persediaan setempat.

Semua pelajaran dalam kursus ini

Menstrukturkan Langkah Transformasi sebagai Fungsi
Memp parameterkan Pipeline dengan Dict Konfigurasi
Menguji Langkah Pipeline dengan Pernyataan
Menjadualkan dan Merekod Pelaksanaan Pipeline

← Kembali ke Pandas & NumPy Academy