การจัดเวลาและบันทึกการทำงานของกระบวนการ
เรียกใช้กระบวนการเป็นสคริปต์ Python จากบรรทัดคำสั่ง บันทึกเวลาเริ่มต้นและสิ้นสุด และใช้ cron หรือตัวจัดตารางเวลาเพื่อทำงานอัตโนมัติ
การจัดเวลาและบันทึกการทำงานของกระบวนการ เป็นบทเรียน Pandas & NumPy Academy ฟรีบน CoddyKit นี่คือบทเรียนที่ 4 จากทั้งหมด 4 บทเรียน คุณสามารถอ่านบทเรียนทั้งหมดด้านล่างฟรี — จากนั้นลองปฏิบัติด้วยตัวคุณเองในเบราว์เซอร์พร้อมตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7 บทเรียนนี้เป็นส่วนหนึ่งของเส้นทางการเรียน Pandas & NumPy Academy และความก้าวหน้าของคุณจะซิงค์ข้ามเว็บและแอป CoddyKit คอร์ส Pandas & NumPy Academy มีบทเรียนทั้งหมด 4 บทเรียน
จาก Notebook สู่สคริปต์
pipeline ที่ทำงานเฉพาะเมื่อผู้พัฒนาเปิด notebook ด้วยตนเองเท่านั้น แทบไม่สร้างคุณค่าทางธุรกิจนอกเหนือจากการทำงานครั้งแรก หากต้องการให้ทำงานโดยอัตโนมัติทุกวัน pipeline ต้องจัดโครงสร้างเป็น สคริปต์ Python ที่เรียกใช้ได้จากบรรทัดคำสั่ง: 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)การตั้งค่าการบันทึกการทำงานของ Python
โมดูล logging ในตัวของ Python เป็นเครื่องมือที่เหมาะสมสำหรับบันทึกการทำงานของ pipeline ไม่ใช่คำสั่ง print() ให้กำหนดค่าตัวบันทึกให้ส่งผลลัพธ์ทั้งไปยังคอนโซลและไฟล์โดยใช้ 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.')การบันทึกเวลาเริ่มต้นและสิ้นสุดของ Pipeline
บันทึกเวลาเริ่มต้น เวลาสิ้นสุด และเวลาที่ใช้ไปของการทำงาน pipeline ทุกครั้ง ข้อมูลนี้เป็นค่าพื้นฐานสำหรับเปรียบเทียบ หากโดยปกติ pipeline ใช้เวลา 45 วินาที แต่วันนี้ใช้เวลา 8 นาที ก็แสดงว่ามีบางอย่างเปลี่ยนแปลง เช่น ไฟล์ข้อมูลเข้าใหญ่ขึ้น 10 เท่า หรือการสืบค้นฐานข้อมูลทำงานช้าลง รายการบันทึกเวลาเริ่มต้นและสิ้นสุดที่มีการระบุเวลาทำให้เปรียบเทียบได้ง่ายจากไฟล์บันทึกเพียงอย่างเดียว
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.')การจัดตารางเวลาด้วย cron บน Linux/Mac
cron เป็นตัวจัดตารางเวลามาตรฐานของ Unix สำหรับงานที่เกิดซ้ำ ให้แก้ไข crontab ด้วย crontab -e และเพิ่มบรรทัดที่ระบุเวลาสำหรับเรียกใช้สคริปต์ รูปแบบคือ: นาที ชั่วโมง วัน เดือน วันในสัปดาห์ คำสั่ง pipeline ที่ต้องทำงานทุกวันเวลา 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 ของ Python
ไลบรารี schedule มอบวิธีแบบ Python ล้วนสำหรับเรียกใช้งานตามช่วงเวลาที่กำหนด โดยไม่ต้องแก้ไข cron มีประโยชน์ในสภาพแวดล้อมที่ไม่มี cron เช่น Windows หรือเมื่อต้องการให้ตรรกะของตัวจัดตารางเวลาอยู่ภายในกระบวนการ Python เอง ให้ครอบ pipeline ไว้ในลูปงานตามตารางเวลา และทำให้กระบวนการทำงานต่อเนื่องเพื่อเรียกใช้ซ้ำ
# 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')การจัดการข้อผิดพลาดและรหัสการจบการทำงาน
สคริปต์ pipeline ควรคืนค่ารหัสการจบการทำงานที่ไม่ใช่ศูนย์เมื่อทำงานล้มเหลว เพื่อให้ตัวจัดตารางเวลาทราบว่างานล้มเหลว ให้ครอบการทำงานหลักด้วยบล็อก try/except และเรียก sys.exit(1) เมื่อเกิดความล้มเหลว cron, Jenkins และ Airflow ล้วนตรวจสอบรหัสการจบการทำงาน รหัสที่ไม่ใช่ศูนย์จะทำให้เกิดการแจ้งเตือน การทำงานซ้ำ หรือการส่งการแจ้งให้ทราบ ข้อยกเว้นที่ไม่ได้จัดการและไม่ได้กำหนดรหัสการจบการทำงานอาจไม่ถูกตรวจพบโดยระบบติดตามอัตโนมัติ
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
หลังการทำงานสำเร็จ ให้เขียนไฟล์สรุป JSON ขนาดเล็กไว้ข้างผลลัพธ์ โดยระบุเวลาที่ทำงาน จำนวนแถวข้อมูลเข้า จำนวนแถวผลลัพธ์ จำนวนแถวที่ถูกลบ และเวลาที่ใช้ ระบบติดตามและแดชบอร์ดสามารถอ่านไฟล์นี้เพื่อติดตามแนวโน้มสถานะของ pipeline ในระยะยาว แดชบอร์ดที่แสดง จำนวนแถวผลลัพธ์ในช่วง 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.')การจัดตารางเวลาแบบทำซ้ำได้: หลีกเลี่ยงการทำงานซ้ำสองครั้ง
หาก pipeline ที่ทำงานตามตารางเวลาถูกเรียกใช้สองครั้งโดยไม่ตั้งใจ pipeline ไม่ควรทำให้ผลลัพธ์เสียหาย ออกแบบขั้นตอนโหลดให้ทำซ้ำได้อย่างปลอดภัย โดยใช้ชื่อไฟล์ผลลัพธ์ที่มีวันที่ หรือเขียนทับผลลัพธ์เดิมด้วยผลลัพธ์ล่าสุด สำหรับการโหลดลงฐานข้อมูล ให้ใช้ 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 Python มี 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.')สคริปต์ไปป์ไลน์ตามกำหนดเวลาฉบับสมบูรณ์
รวมองค์ประกอบทั้งหมด ได้แก่ การแยกวิเคราะห์อาร์กิวเมนต์ การกำหนดค่า logging สรุปการทำงาน การจัดการข้อผิดพลาด และรหัสการออก ให้เป็นสคริปต์ไปป์ไลน์ฉบับสมบูรณ์ สคริปต์นี้สามารถนำไปใช้ในสภาพแวดล้อมใดก็ได้ ระบุไฟล์การกำหนดค่า และตั้งเวลาให้ทำงานด้วย cron หรือเครื่องมือจัดการเวิร์กโฟลว์ใด ๆ ได้ ทุกครั้งที่ทำงาน สคริปต์จะสร้างไฟล์บันทึกที่มีวันที่ ไฟล์สรุปการทำงาน และไฟล์ผลลัพธ์ที่มีวันที่ ทำให้สามารถตรวจสอบย้อนกลับและทำซ้ำการทำงานแต่ละครั้งได้อย่างอิสระอย่างครบถ้วน
# 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')ตรวจสอบความเข้าใจอย่างรวดเร็ว
ทดสอบความเข้าใจแนวคิดด้านการวิเคราะห์ข้อมูลจากบทเรียนนี้
สรุปบทเรียน
ในบทเรียนนี้ คุณได้เรียนรู้เกี่ยวกับ การจัดโครงสร้างไปป์ไลน์เป็นสคริปต์บรรทัดคำสั่งพร้อมการแยกวิเคราะห์อาร์กิวเมนต์และ logging การตั้งเวลาทำงานด้วย cron และการจัดการความล้มเหลวด้วยรหัสการออกที่ไม่เป็นศูนย์และการแจ้งเตือน รวมถึง การเขียนไฟล์สรุปการทำงานและการออกแบบขั้นตอนโหลดแบบทำซ้ำได้เพื่อให้การทำงานอัตโนมัติมีความน่าเชื่อถือ ขอแสดงความยินดีที่เรียนจบแทร็กการวิเคราะห์ข้อมูล: Pandas และ NumPy
เรียนรู้ Python ด้วย AI tutor — ฟรี
เขียนและเรียกใช้โค้ดจริงในเบราว์เซอร์ของคุณ รับความช่วยเหลือทันทีจาก AI tutor 24/7 และเรียนรู้ต่อจากที่คุณหยุดบนเว็บหรือในแอป
- คอร์ส
- 30
- บทเรียน
- 120
คำถามที่พบบ่อย
บทเรียน “การจัดเวลาและบันทึกการทำงานของกระบวนการ” ฟรีหรือไม่
ใช่ — ข้อความเต็มของ “การจัดเวลาและบันทึกการทำงานของกระบวนการ” ฟรีให้อ่านที่นี่บนเว็บ เพื่อปฏิบัติแบบโต้ตอบ (ตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7) และปลดล็อคส่วนที่เหลือของคอร์ส Pandas & NumPy Academy ให้อัปเกรดเป็น CoddyKit PRO คอร์ส Pandas & NumPy Academy มีบทเรียนทั้งหมด 4 บทเรียน
คุณจะเรียนรู้อะไรในบทเรียน “การจัดเวลาและบันทึกการทำงานของกระบวนการ”
เรียกใช้กระบวนการเป็นสคริปต์ Python จากบรรทัดคำสั่ง บันทึกเวลาเริ่มต้นและสิ้นสุด และใช้ cron หรือตัวจัดตารางเวลาเพื่อทำงานอัตโนมัติ คุณปฏิบัติ Pandas & NumPy Academy ด้วยโค้ดที่ใช้งานได้จริงที่คุณเรียกใช้โดยตรงในเบราว์เซอร์ และติวเตอร์ AI ตลอด 24/7 ตอบคำถามของคุณขณะที่คุณไปผ่านบทเรียน
คุณต้องมีประสบการณ์ก่อนที่จะเริ่มเรียน Pandas & NumPy Academy หรือไม่
ไม่จำเป็นต้องมีประสบการณ์มาก่อน Pandas & NumPy Academy บน CoddyKit ออกแบบมาสำหรับผู้เริ่มต้นไปจนถึงผู้เรียนขั้นสูง คุณสามารถเริ่มต้นที่นี่หรือเริ่มจากตัวแรกและเรียนด้วยความเร็วของคุณเอง นี่คือบทเรียนที่ 4 จากทั้งหมด 4 บทเรียน
บทเรียน “การจัดเวลาและบันทึกการทำงานของกระบวนการ” ใช้เวลานานแค่ไหน
บทเรียน CoddyKit ส่วนใหญ่ใช้เวลาประมาณ 5–10 นาที แต่ละบทเรียนจึงสั้นและเป็นแบบโต้ตอบ คุณสามารถก้าวหน้าอย่างต่อเนื่องและกลับมาเรียนต่อจากตรงที่เพิ่งหยุดบนเว็บและแอปได้เลย
ฉันเขียนและรันโค้ดในบทเรียน Pandas & NumPy Academy นี้ได้ไหม
ได้ บทเรียน Pandas & NumPy Academy ทุกบทมีตัวแก้ไขโค้ดในตัว คุณจึงเขียนและรันโค้ดจริงได้เลยในเบราว์เซอร์ และได้รับข้อเสนอแนะจาก AI ในทันที — ไม่ต้องติดตั้งในเครื่องของคุณ
บทเรียนทั้งหมดในหลักสูตรนี้
- การจัดโครงสร้างขั้นตอนการแปลงเป็นฟังก์ชัน
- การกำหนดพารามิเตอร์ให้กระบวนการด้วยพจนานุกรมการตั้งค่า
- การทดสอบขั้นตอนกระบวนการด้วยการยืนยันเงื่อนไข
- การจัดเวลาและบันทึกการทำงานของกระบวนการ