การต่อเมธอดด้วย pipe()
เขียนกระบวนการแปลงข้อมูลที่อ่านง่ายด้วย pipe() เพื่อเชื่อมฟังก์ชันกำหนดเองเข้ากับเมธอดดั้งเดิมของ Pandas
การต่อเมธอดด้วย pipe() เป็นบทเรียน Pandas & NumPy Academy ฟรีบน CoddyKit นี่คือบทเรียนที่ 4 จากทั้งหมด 4 บทเรียน คุณสามารถอ่านบทเรียนทั้งหมดด้านล่างฟรี — จากนั้นลองปฏิบัติด้วยตัวคุณเองในเบราว์เซอร์พร้อมตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7 บทเรียนนี้เป็นส่วนหนึ่งของเส้นทางการเรียน Pandas & NumPy Academy และความก้าวหน้าของคุณจะซิงค์ข้ามเว็บและแอป CoddyKit คอร์ส Pandas & NumPy Academy มีบทเรียนทั้งหมด 4 บทเรียน
ปัญหาของตัวแปรระหว่างทาง
pipeline สำหรับทำความสะอาดข้อมูลที่ไม่ใช้ pipe() มักสะสมตัวแปรระหว่างทางไว้มากมาย เช่น df1 = clean(df), df2 = transform(df1), df3 = enrich(df2) ตัวแปรเหล่านี้ทำให้เนมสเปซรกรุงรัง ทำให้การแก้จุดบกพร่องยากขึ้น และล่อลวงให้นักพัฒนาใช้งานซ้ำอย่างไม่ถูกต้อง ผลลัพธ์คือโค้ดที่อ่านจากบนลงล่างได้ยากในฐานะลำดับการแปลงข้อมูล
import pandas as pd
df = pd.read_csv('orders.csv')
# Without pipe — intermediate variables everywhere
df1 = df.dropna(subset=['revenue'])
df2 = df1[df1['quantity'] > 0]
df3 = df2.assign(revenue_per_unit=df2['revenue'] / df2['quantity'])
print(df3.shape)แนะนำ pipe()
DataFrame.pipe(func) จะเรียก func(df) และคืนผลลัพธ์ ทำให้คุณต่อฟังก์ชันแบบกำหนดเองได้ในลักษณะเดียวกับการต่อเมธอดใน Pandas เช่น .dropna().query() ประโยชน์สำคัญคือทุกขั้นตอนการแปลงข้อมูลมีความชัดเจนและอ่านจากซ้ายไปขวาได้ง่าย (หรือจากบนลงล่างเมื่อจัดรูปแบบด้วยวงเล็บ) ซึ่งสอดคล้องกับลำดับตรรกะของ pipeline
def drop_nulls(df):
return df.dropna(subset=['revenue'])
def filter_positive_qty(df):
return df[df['quantity'] > 0]
def add_revenue_per_unit(df):
return df.assign(revenue_per_unit=df['revenue'] / df['quantity'])
# With pipe — clean chain
df_clean = (df
.pipe(drop_nulls)
.pipe(filter_positive_qty)
.pipe(add_revenue_per_unit)
)
print(df_clean.shape)การส่งอาร์กิวเมนต์ผ่าน pipe()
ส่งอาร์กิวเมนต์เพิ่มเติมไปยังฟังก์ชันที่ต่อด้วยอาร์กิวเมนต์แบบระบุชื่อหลังชื่อฟังก์ชัน: df.pipe(func, arg1=val1) รูปแบบฟังก์ชันต้องรับ df เป็นพารามิเตอร์แรก ฟังก์ชันที่รับพารามิเตอร์ทำให้ pipeline กำหนดค่าได้ คุณสามารถเปลี่ยนค่าขีดจำกัด ชื่อคอลัมน์ หรือพฤติกรรมได้โดยไม่ต้องแก้เนื้อหาของฟังก์ชัน เพียงเปลี่ยนอาร์กิวเมนต์ในการเรียกใช้ pipe
def filter_by_region(df, regions):
return df[df['region'].isin(regions)]
def cap_revenue(df, upper):
df = df.copy()
df['revenue'] = df['revenue'].clip(upper=upper)
return df
df_result = (df
.pipe(filter_by_region, regions=['North', 'East'])
.pipe(cap_revenue, upper=1000)
)
print(df_result.shape)การใช้ pipe() ร่วมกับเมธอดในตัว
จุดเด่นของ pipe() คือการผสานการทำงานกับเมธอดดั้งเดิมของ Pandas ในสายคำสั่งเดียวกันได้อย่างราบรื่น คุณสามารถผสม .dropna(), .query(), .rename() และ .pipe(custom_func) ในลำดับใดก็ได้ ทำให้สายคำสั่งทั้งกระชับ (ใช้เมธอดในตัวเมื่อทำได้) และยืดหยุ่น (ใช้ฟังก์ชันแบบกำหนดเองเมื่อเมธอดในตัวไม่เพียงพอ)
df_result = (
df
.dropna(subset=['revenue', 'order_date'])
.query('quantity > 0')
.rename(columns={'unit_price': 'price'})
.pipe(add_revenue_per_unit)
.reset_index(drop=True)
)
print(df_result.head())การแก้จุดบกพร่องของสายคำสั่ง pipe()
การแก้จุดบกพร่องของสายคำสั่งที่ยาวอาจทำได้ยาก เพราะคุณไม่สามารถตรวจสอบสถานะระหว่างทางด้วยการเพิ่มคำสั่งพิมพ์ผลไว้ตรงกลางได้ วิธีหนึ่งคือเขียน ฟังก์ชันแก้จุดบกพร่องแบบส่งต่อ ที่พิมพ์ข้อมูลรูปร่างและคอลัมน์ แล้วคืนค่า DataFrame เดิมโดยไม่เปลี่ยนแปลง คุณสามารถแทรกฟังก์ชันนี้ไว้ที่จุดใดก็ได้ในสายคำสั่ง เพื่อตรวจสอบสถานะ ณ ขั้นตอนนั้นโดยไม่ทำให้สายคำสั่งขาดตอน
def debug(df, label=''):
print(f'[{label}] shape: {df.shape}')
print(f'[{label}] columns: {df.columns.tolist()}')
return df
df_result = (
df
.pipe(drop_nulls)
.pipe(debug, label='after drop_nulls')
.pipe(filter_positive_qty)
.pipe(debug, label='after filter')
)
print('Done')การสร้างสายงานทำความสะอาดข้อมูลแบบสมบูรณ์
รวมขั้นตอนการทำความสะอาดทั้งหมดไว้ในฟังก์ชันสายงานเดียวโดยใช้ pipe() การห่อสายคำสั่งไว้ในฟังก์ชันชื่อ clean_pipeline(df) ทำให้สามารถทดสอบสายงานนี้เป็นหน่วยเดียวได้ เรียกใช้ด้วย DataFrame ดิบ แล้วรับ DataFrame ที่สะอาดกลับมา รูปแบบนี้สอดคล้องกับกระบวนทัศน์ ETL (สกัด แปลง และโหลด) ที่ใช้ในงานวิศวกรรมข้อมูลจริง
def clean_pipeline(df):
return (
df
.dropna(subset=['order_id', 'revenue'])
.drop_duplicates(subset=['order_id'])
.query('quantity > 0 and revenue >= 0')
.pipe(add_revenue_per_unit)
.reset_index(drop=True)
)
df_clean = clean_pipeline(df)
print('Clean rows:', len(df_clean))ความแตกต่างสำคัญระหว่าง pipe() กับ apply()
pipe(func) ส่ง DataFrame ทั้งหมดไปยัง func และคาดว่าจะได้รับ DataFrame (หรือออบเจ็กต์ที่ผ่านการแปลงแล้ว) กลับมา ส่วน apply(func, axis=1) จะส่งข้อมูลทีละแถว ใช้ pipe() สำหรับการแปลง DataFrame ทั้งหมดที่ยังคงรูปร่างเดิม (หรือเปลี่ยนรูปร่างโดยตั้งใจ) และใช้ apply() สำหรับการคำนวณระดับแถวหรือคอลัมน์ ทั้งสองอย่างเสริมกัน ไม่ได้แข่งขันกัน
# pipe: receives the whole DataFrame
def scale_revenue(df, factor=1.0):
df = df.copy()
df['revenue'] = df['revenue'] * factor
return df
# apply: receives one row at a time
df['revenue_x2'] = df.apply(lambda row: row['revenue'] * 2, axis=1)
df_scaled = df.pipe(scale_revenue, factor=1.1)
print('pipe scales all rows at once; apply does row-by-row')ส่วนประกอบสายงานที่นำกลับมาใช้ซ้ำได้
เขียนแต่ละขั้นตอนของสายงานเป็น ฟังก์ชันบริสุทธิ์ — ไม่มีสถานะส่วนกลาง รับ DataFrame เข้ามา และคืนค่า DataFrame ออกไป ฟังก์ชันบริสุทธิ์ทดสอบแบบหน่วยได้ง่าย เพียงเรียกใช้ด้วย DataFrame ขนาดเล็กสำหรับทดสอบ แล้วตรวจสอบรูปร่างผลลัพธ์และค่าคอลัมน์ ไลบรารีฟังก์ชันสายงานที่ผ่านการทดสอบและนำกลับมาใช้ซ้ำได้จะช่วยเร่งการวิเคราะห์ชุดข้อมูลใหม่ที่มีข้อกำหนดด้านการทำความสะอาดคล้ายกันได้อย่างมาก
def normalise_strings(df, cols):
df = df.copy()
for col in cols:
df[col] = df[col].str.strip().str.lower()
return df
def parse_dates(df, cols):
df = df.copy()
for col in cols:
df[col] = pd.to_datetime(df[col])
return df
df_result = (
df
.pipe(normalise_strings, cols=['region', 'category'])
.pipe(parse_dates, cols=['order_date'])
)
print(df_result.dtypes)การบันทึกขั้นตอนของสายงานด้วย pipe()
เพิ่มการบันทึกข้อมูลอย่างเป็นโครงสร้างภายในแต่ละฟังก์ชันของสายงาน เพื่อให้ตรวจสอบการแปลงข้อมูลทุกครั้งในระบบจริงได้ บันทึกจำนวนแถวขาเข้า จำนวนแถวขาออก และสถิติที่เกี่ยวข้อง (เช่น จำนวนแถวที่ถูกลบโดยตัวกรอง) วิธีนี้ทำให้มีประวัติการทำงานครบถ้วนของการเรียกใช้สายงานแต่ละครั้ง โดยไม่ต้องใช้ตัวจัดการกระบวนงานภายนอกสำหรับบันทึกการตรวจสอบพื้นฐาน
import logging
logging.basicConfig(level=logging.INFO)
def logged_dropna(df, subset):
before = len(df)
df = df.dropna(subset=subset)
after = len(df)
logging.info(f'dropna: {before - after} rows removed, {after} remaining')
return df
df_clean = df.pipe(logged_dropna, subset=['revenue'])
print('Logged pipeline complete.')ขั้นตอนแบบมีเงื่อนไขในสายงาน
บางครั้งขั้นตอนการทำความสะอาดควรทำงานก็ต่อเมื่อมีค่าสถานะที่กำหนดหรือเมื่อเนื้อหาข้อมูลตรงตามเงื่อนไข คุณสามารถเพิ่มขั้นตอนแบบมีเงื่อนไขในสายคำสั่ง pipe ได้โดยแทรกฟังก์ชันที่เลือกว่าจะคงเดิมหรือแปลงข้อมูล ฟังก์ชันนี้จะตรวจสอบเงื่อนไข แล้วใช้การแปลงหรือคืนค่า DataFrame เดิมโดยไม่เปลี่ยนแปลง วิธีนี้ช่วยคงโครงสร้างของสายคำสั่งไว้ พร้อมรองรับขั้นตอนที่เลือกทำได้
def maybe_cap_revenue(df, cap=None):
if cap is None:
return df
df = df.copy()
df['revenue'] = df['revenue'].clip(upper=cap)
return df
CAPPING_ENABLED = True
CAP_VALUE = 1000 if CAPPING_ENABLED else None
df_result = df.pipe(maybe_cap_revenue, cap=CAP_VALUE)
print('Conditional step applied:', CAPPING_ENABLED)การส่งออกผลลัพธ์ของสายงาน
ขั้นตอนสุดท้ายของสายคำสั่ง pipe() มักเป็นการส่งออกข้อมูล คุณสามารถเชื่อมฟังก์ชันส่งออกแบบกำหนดเองด้วย pipe() หรือเรียกใช้เมธอดส่งออกดั้งเดิมของ Pandas หลังจบสายคำสั่งก็ได้ การใช้ขั้นตอน pipe(save_to_parquet) ทำให้ส่วนการส่งออกเป็นส่วนหนึ่งของเอกสารสายคำสั่ง และรับประกันว่าจะทำงานกับ DataFrame สุดท้ายที่ทำความสะอาดแล้ว ไม่ใช่เวอร์ชันระหว่างทาง
def save_parquet(df, path):
df.to_parquet(path, index=False)
print(f'Saved {len(df)} rows to {path}')
return df # return df so the chain can continue if needed
df_final = (
df
.pipe(clean_pipeline)
.pipe(save_parquet, path='orders_final.parquet')
)
print('Pipeline complete.')ตรวจสอบความเข้าใจอย่างรวดเร็ว
ทดสอบความเข้าใจแนวคิดด้านการวิเคราะห์ข้อมูลจากบทเรียนนี้
สรุปบทเรียน
ในบทเรียนนี้ คุณได้เรียนรู้เรื่อง การใช้ pipe() เพื่อเชื่อมฟังก์ชันแบบกำหนดเองเข้ากับเมธอดดั้งเดิมของ Pandas การสร้างขั้นตอนสายงานที่นำกลับมาใช้ซ้ำได้ รับพารามิเตอร์ และทดสอบได้ในรูปฟังก์ชันบริสุทธิ์ และ การเพิ่มบันทึกการแก้จุดบกพร่องและขั้นตอนแบบมีเงื่อนไขภายในสายคำสั่ง pipe บทถัดไป เราจะสำรวจการจัดโครงสร้างขั้นตอนการแปลงข้อมูลให้อยู่ในรูปฟังก์ชันสำหรับสายงาน ETL ระดับใช้งานจริง
คำถามที่พบบ่อย
บทเรียน “การต่อเมธอดด้วย pipe()” ฟรีหรือไม่
ใช่ — ข้อความเต็มของ “การต่อเมธอดด้วย pipe()” ฟรีให้อ่านที่นี่บนเว็บ เพื่อปฏิบัติแบบโต้ตอบ (ตัวแก้ไขโค้ดในตัวและติวเตอร์ AI ตลอด 24/7) และปลดล็อคส่วนที่เหลือของคอร์ส Pandas & NumPy Academy ให้อัปเกรดเป็น CoddyKit PRO คอร์ส Pandas & NumPy Academy มีบทเรียนทั้งหมด 4 บทเรียน
คุณจะเรียนรู้อะไรในบทเรียน “การต่อเมธอดด้วย pipe()”
เขียนกระบวนการแปลงข้อมูลที่อ่านง่ายด้วย pipe() เพื่อเชื่อมฟังก์ชันกำหนดเองเข้ากับเมธอดดั้งเดิมของ Pandas คุณปฏิบัติ Pandas & NumPy Academy ด้วยโค้ดที่ใช้งานได้จริงที่คุณเรียกใช้โดยตรงในเบราว์เซอร์ และติวเตอร์ AI ตลอด 24/7 ตอบคำถามของคุณขณะที่คุณไปผ่านบทเรียน
คุณต้องมีประสบการณ์ก่อนที่จะเริ่มเรียน Pandas & NumPy Academy หรือไม่
ไม่จำเป็นต้องมีประสบการณ์มาก่อน Pandas & NumPy Academy บน CoddyKit ออกแบบมาสำหรับผู้เริ่มต้นไปจนถึงผู้เรียนขั้นสูง คุณสามารถเริ่มต้นที่นี่หรือเริ่มจากตัวแรกและเรียนด้วยความเร็วของคุณเอง นี่คือบทเรียนที่ 4 จากทั้งหมด 4 บทเรียน
บทเรียน “การต่อเมธอดด้วย pipe()” ใช้เวลานานแค่ไหน
บทเรียน CoddyKit ส่วนใหญ่ใช้เวลาประมาณ 5–10 นาที แต่ละบทเรียนจึงสั้นและเป็นแบบโต้ตอบ คุณสามารถก้าวหน้าอย่างต่อเนื่องและกลับมาเรียนต่อจากตรงที่เพิ่งหยุดบนเว็บและแอปได้เลย
ฉันเขียนและรันโค้ดในบทเรียน Pandas & NumPy Academy นี้ได้ไหม
ได้ บทเรียน Pandas & NumPy Academy ทุกบทมีตัวแก้ไขโค้ดในตัว คุณจึงเขียนและรันโค้ดจริงได้เลยในเบราว์เซอร์ และได้รับข้อเสนอแนะจาก AI ในทันที — ไม่ต้องติดตั้งในเครื่องของคุณ
บทเรียนทั้งหมดในหลักสูตรนี้
- apply() กับคอลัมน์และแถว
- apply() กับ GroupBy
- map() และ applymap() สำหรับการดำเนินการทีละองค์ประกอบ
- การต่อเมธอดด้วย pipe()