Pandas & NumPy Academy · Pelajaran

Menjadwalkan dan Mencatat Proses Pipeline

Jalankan pipeline sebagai skrip Python dari baris perintah, catat waktu mulai dan selesai, lalu gunakan cron atau penjadwal untuk otomatisasi.

Menjadwalkan dan Mencatat Proses Pipeline adalah pelajaran Pandas & NumPy Academy gratis di CoddyKit. Ini adalah pelajaran 4 dari 4. Kamu bisa membaca pelajaran lengkapnya di bawah secara gratis — lalu praktikkan langsung di browser dengan editor kode bawaan dan tutor AI 24/7. Ini adalah bagian dari jalur belajar Pandas & NumPy Academy, dan progresmu tersinkronisasi di web dan aplikasi CoddyKit. Kursus Pandas & NumPy Academy mencakup 4 pelajaran total.

Dari Notebook ke Skrip

Alur data yang hanya berjalan ketika pengembang membuka notebook secara manual tidak memberikan nilai bisnis setelah proses pertama. Agar dapat berjalan otomatis setiap hari, alur data harus disusun sebagai skrip Python yang dapat dijalankan dari baris perintah: python pipeline.py. Hal ini memerlukan titik masuk if __name__ == '__main__':, penguraian argumen baris perintah, dan pencatatan yang benar—tiga pilar skrip untuk 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)

Mengonfigurasi Pencatatan Python

Modul logging bawaan Python adalah alat yang tepat untuk log alur data—bukan pernyataan print(). Konfigurasikan pencatat dengan output konsol dan output file menggunakan logging.basicConfig(). Catat pada tingkat INFO untuk kemajuan normal dan ERROR untuk kegagalan. Log berbasis file tetap tersedia setelah proses berakhir, yang sangat penting untuk men-debug eksekusi terjadwal yang tidak diawasi siapa pun.

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

Mencatat Awal dan Akhir Alur Data

Selalu catat waktu mulai, waktu selesai, dan waktu yang berlalu dari eksekusi alur data. Hal ini menjadi dasar perbandingan: jika alur data biasanya berjalan selama 45 detik tetapi hari ini memerlukan 8 menit, berarti ada sesuatu yang berubah—mungkin file input berukuran 10 kali lebih besar atau kueri basis data berjalan lambat. Entri log waktu mulai dan selesai yang memiliki stempel waktu membuat perbandingan ini mudah dilakukan hanya dari file 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

Mencatat Jumlah Baris per Tahap

Catat jumlah baris yang masuk dan keluar dari setiap tahap transformasi. Log yang jelas akan terlihat seperti: extract: 50,000 rows → drop_nulls: 49,200 rows → filter: 47,800 rows → output: 47,800 rows. Jejak ini segera memperjelas berapa banyak baris yang dihapus pada setiap tahap dan apakah jumlahnya sesuai harapan. Penghapusan yang tidak wajar akan terlihat sebagai kesenjangan pada jumlah yang dicatat.

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

Penjadwalan dengan cron di Linux/Mac

cron adalah penjadwal Unix standar untuk pekerjaan berulang. Edit crontab dengan crontab -e dan tambahkan baris yang menentukan kapan skrip dijalankan. Formatnya adalah: menit jam hari bulan hari-dalam-minggu perintah. Alur data yang harus berjalan setiap hari pada pukul 06.00 menggunakan 0 6 * * * /usr/bin/python /path/to/pipeline.py. Selalu gunakan jalur absolut dalam entri cron karena cron berjalan dalam lingkungan minimal tanpa pengaturan PATH dari 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')

Penjadwalan dengan Pustaka schedule Python

Pustaka schedule menyediakan cara berbasis Python murni untuk menjalankan pekerjaan pada interval tertentu tanpa menggunakan cron. Pustaka ini berguna di lingkungan yang tidak menyediakan cron (Windows) atau ketika Anda ingin logika penjadwalan berada di dalam proses Python itu sendiri. Bungkus alur data dalam loop pekerjaan terjadwal dan pertahankan proses tetap berjalan agar pekerjaan dapat dieksekusi 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')

Penanganan Error dan Kode Keluar

Skrip alur data harus mengembalikan kode keluar bukan nol ketika gagal agar penjadwal mengetahui bahwa pekerjaan tersebut gagal. Bungkus eksekusi utama dalam blok try/except dan panggil sys.exit(1) saat terjadi kegagalan. cron, Jenkins, dan Airflow semuanya memeriksa kode keluar: kode bukan nol memicu peringatan, eksekusi ulang, atau notifikasi. Pengecualian yang tidak tertangani dan tidak menetapkan kode keluar dapat luput dari pemantauan otomatis.

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 File Ringkasan Eksekusi Alur Data

Setelah eksekusi berhasil, tulis file ringkasan JSON kecil bersama output. Sertakan stempel waktu eksekusi, jumlah baris input, jumlah baris output, jumlah baris yang dihapus, dan waktu yang berlalu. Sistem pemantauan dan dasbor dapat membaca file ini untuk melacak tren kesehatan alur data dari waktu ke waktu. Dasbor yang menampilkan jumlah baris output selama 30 hari terakhir memudahkan Anda menemukan hari ketika sumber data mulai mengirimkan lebih sedikit rekaman.

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

Penjadwalan Idempoten: Menghindari Eksekusi Ganda

Jika alur data terjadwal tidak sengaja dipicu dua kali, alur tersebut tidak boleh merusak output. Rancang tahap pemuatan agar idempoten: gunakan nama file output bertanggal, atau timpa output yang sama dengan hasil terbaru. Untuk pemuatan ke basis data, gunakan if_exists='replace' atau pola UPSERT. Jangan pernah menggunakan mode append tanpa tahap deduplikasi, karena setiap eksekusi terjadwal akan menambahkan baris duplikat ke tabel 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 Peringatan saat Pipeline Gagal

Untuk pipeline yang menjadi tumpuan operasional bisnis, tidak adanya kabar setelah terjadi kegagalan sangat berbahaya. Siapkan peringatan sederhana: jika file ringkasan proses tidak diperbarui dalam rentang waktu yang diharapkan, kirim email atau pesan Slack. smtplib Python dapat mengirim email saat terjadi kegagalan, atau Anda dapat menggunakan webhook untuk mengirim pesan ke Slack. Berikan peringatan segera saat kode keluar 1 atau 2 agar analis mengetahui bahwa pembaruan harian terlewat sebelum pihak bisnis menyadarinya.

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 Terjadwal yang Lengkap

Gabungkan semua bagian — penguraian argumen, konfigurasi pencatatan, ringkasan proses, penanganan kesalahan, dan kode keluar — menjadi skrip pipeline yang lengkap. Skrip ini dapat ditempatkan di lingkungan apa pun, diarahkan ke file konfigurasi, lalu dijadwalkan dengan cron atau pengatur alur kerja apa pun. Pada setiap eksekusi, skrip ini menghasilkan file log bertanggal, ringkasan proses, dan file keluaran bertanggal, sehingga setiap proses dapat diaudit sepenuhnya dan direproduksi secara independen.

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

Pemeriksaan Singkat

Uji pemahaman Anda tentang konsep Analisis Data dari pelajaran ini.

Ringkasan Pelajaran

Dalam pelajaran ini, Anda mempelajari: menyusun pipeline sebagai skrip baris perintah dengan penguraian argumen dan pencatatan, menjadwalkan dengan cron serta menangani kegagalan menggunakan kode keluar bukan nol dan peringatan, serta menulis file ringkasan proses dan merancang langkah pemuatan idempoten untuk eksekusi otomatis yang andal. Selamat karena telah menyelesaikan jalur Analisis Data: Pandas dan NumPy!

Pertanyaan yang Sering Diajukan

Apakah pelajaran “Menjadwalkan dan Mencatat Proses Pipeline” gratis?

Ya — teks lengkap “Menjadwalkan dan Mencatat Proses Pipeline” gratis dibaca di sini di web. Untuk praktiknya secara interaktif (editor kode bawaan dan tutor AI 24/7) dan buka sisa kursus Pandas & NumPy Academy, upgrade ke CoddyKit PRO. Kursus Pandas & NumPy Academy mencakup 4 pelajaran total.

Apa yang akan aku pelajari di “Menjadwalkan dan Mencatat Proses Pipeline”?

Jalankan pipeline sebagai skrip Python dari baris perintah, catat waktu mulai dan selesai, lalu gunakan cron atau penjadwal untuk otomatisasi. Kamu berlatih Pandas & NumPy Academy dengan kode praktik yang langsung kamu jalankan di browser, dan tutor AI 24/7 menjawab pertanyaanmu saat kamu mengerjakan pelajaran ini.

Apakah aku perlu pengalaman untuk memulai Pandas & NumPy Academy?

Tidak diperlukan pengalaman sebelumnya. Pandas & NumPy Academy di CoddyKit dirancang untuk pemula hingga pelajar tingkat lanjut, jadi kamu bisa memulai di sini atau dari awal dan belajar sesuai kecepatan kamu sendiri. Ini adalah pelajaran 4 dari 4.

Berapa lama pelajaran “Menjadwalkan dan Mencatat Proses Pipeline” memakan waktu?

Sebagian besar pelajaran CoddyKit memakan waktu sekitar 5–10 menit. Setiap pelajaran ringkas dan interaktif, jadi kamu membuat kemajuan stabil dan melanjutkan dari tempat kamu tinggalkan di web dan aplikasi.

Bisakah aku menulis dan menjalankan kode dalam pelajaran Pandas & NumPy Academy ini?

Ya. Setiap pelajaran Pandas & NumPy Academy menyertakan editor kode bawaan, jadi kamu menulis dan menjalankan kode nyata langsung di browser dan mendapatkan umpan balik AI instan — tidak diperlukan penyiapan lokal.

Semua pelajaran dalam kursus ini

Menyusun Langkah Transformasi sebagai Fungsi
Membuat Pipeline Berparameter dengan Dict Konfigurasi
Menguji Langkah Pipeline dengan Assertion
Menjadwalkan dan Mencatat Proses Pipeline

← Kembali ke Pandas & NumPy Academy