毎週月曜の、同じ掃除
店長は、毎週月曜に新しい書き出しファイルを送ります。先週と同じ傷があります。重複、数字のはずが文字、おかしな表記です。
毎回、ノートブックで手作業で直すこともできます。あるいは、手順を順番どおりに1度だけスクリプトへ書き、毎回同じように動かせます。
このレッスンのゴール
メモ
掃除し、結合し、報告し、確かめに失敗したら止まる小さなパイプラインを書ける。
キーワード
| ことば | いみ | れい |
|---|---|---|
| パイプライン | 段階を順に動かし、各段階が前の結果を使うしくみ | |
| 段階 | パイプラインの1つの手順 | 重複を除く |
| 確かめ | パイプラインの中で、全体を止められるテスト | |
| 終了コード | スクリプトが終わるときの数。成功なら 0 | sys.exit(1) |
| 再現できる | 動かすたびに、同じ結果が出ること |
学生が入力するセルは、このプレビューでは空白です。
決まった順番の段階
パイプラインは、段階のリストです。各段階は表を受け取り、表を返します。順番が大事です。日付がきれいになる前に、日付で結合はできません。
各段階のあとに、残った行数を表示できます。数が急に減れば、どこかの手順がうまくいかなかった最初の合図です。
例: 今日のパイプラインの段階
| 段階 | することと | 期待する行数 |
|---|---|---|
| 読み込み | 汚れた書き出しファイルを読む | 367 |
| 重複なし | 2回ある行を除く | 365 |
| 本当の台数 | 0以上の台数を、数として残す | 363 |
| 結合 | 祝日の列を足し、どの日も残す | 363 |
学生が入力するセルは、このプレビューでは空白です。
祝日の一覧をもう一度書く
docker exec jupyter sh -c "printf '%s\n' 'date,holiday' '2025-01-01,元日' '2025-01-13,成人の日' '2025-02-11,建国記念の日' '2025-03-20,春分の日' '2025-04-29,昭和の日' '2025-05-05,こどもの日' '2025-07-21,海の日' '2025-11-03,文化の日' > holidays.csv"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
スクリプトは、約20行
次のブロックがスクリプトです。まず読んでください。関数 stage は、残った行数を表示し、least より少なければ、終了コード 1 でスクリプト全体を止めます。
数 least は、動かすときに渡します。確かめをどれだけ厳しくするかは、あなたが決めます。
steps/pipeline.py
import sys
import pandas as pd
from mess import make
least = int(sys.argv[1])
def stage(name, table):
print(name, "->", len(table), "行")
if len(table) < least:
print("中止:", name, "のあと", least, "行より少ない")
sys.exit(1)
return table
make()
table = stage("読み込み", pd.read_csv("messy.csv"))
table = stage("重複なし", table.drop_duplicates())
table["hires"] = pd.to_numeric(table["hires"], errors="coerce")
table = stage("本当の台数", table[table["hires"] >= 0])
table["weather"] = table["weather"].str.strip().str.lower()
table = stage("結合", table.merge(pd.read_csv("holidays.csv"), on="date", how="left"))
table.to_csv("report.csv", index=False)
print(table.groupby("weather")["hires"].mean().round(1).to_string())
パイプラインをラボに書く
docker exec jupyter sh -c "printf '%s\n' 'import sys' 'import pandas as pd' 'from mess import make' '' 'least = int(sys.argv[1])' '' 'def stage(name, table):' ' print(name, \"->\", len(table), \"行\")' ' if len(table) < least:' ' print(\"中止:\", name, \"のあと\", least, \"行より少ない\")' ' sys.exit(1)' ' return table' '' 'make()' 'table = stage(\"読み込み\", pd.read_csv(\"messy.csv\"))' 'table = stage(\"重複なし\", table.drop_duplicates())' 'table[\"hires\"] = pd.to_numeric(table[\"hires\"], errors=\"coerce\")' 'table = stage(\"本当の台数\", table[table[\"hires\"] >= 0])' 'table[\"weather\"] = table[\"weather\"].str.strip().str.lower()' 'table = stage(\"結合\", table.merge(pd.read_csv(\"holidays.csv\"), on=\"date\", how=\"left\"))' 'table.to_csv(\"report.csv\", index=False)' 'print(table.groupby(\"weather\")[\"hires\"].mean().round(1).to_string())' > steps/pipeline.py"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
下限300行で動かす
docker exec jupyter python3 steps/pipeline.py 300ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
パイプラインを止める確かめ
確かめは、「だめだ」と言えるときだけ役に立ちます。ここでは、残った行数の下限です。データが満たせない下限を渡すと、それを破る最初の段階で、パイプラインが止まります。
スクリプトは終了コード 1 で終わります。動かした側は、失敗したと分かります。
365行以上を求める
docker exec jupyter python3 steps/pipeline.py 365 || echo "終了コード $?"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
400行以上を求める
docker exec jupyter python3 steps/pipeline.py 400 || echo "終了コード $?"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
終わった実行が書いたものを確かめる
docker exec jupyter python3 -c "import pandas as pd; r=pd.read_csv('report.csv'); print(r.shape, r['holiday'].notna().sum(), r['date'].is_unique, r['hires'].isna().sum())"ここでは表示のみで、実行されません。無料アカウントを作成すると、実行ボタンで自分のラボのターミナルに入力できます。
よくある間違い
注意
エラーなしで終わったパイプラインは正しい、と思いがちです。通ったのは、誰かが書いた確かめだけです。行数は、天気の語の書き方がそろったかを教えません。見た傷ごとに、確かめを足しましょう。
段階のひな形を Python で
# 段階のひな形を、ふつうの Python で。やって、知らせて、残りが少なすぎたら止めます。
rows = [3, 8, 1, 9, 4, 2]
least = 3
def keep_above(rows, limit):
kept = []
for one in rows:
if one > limit:
kept.append(one)
return kept
def stage(name, rows, limit):
rows = keep_above(rows, limit)
print(name, "->", len(rows), "行")
if len(rows) < least:
print("中止:", name)
return None
return rows
rows = stage("2より大きいものだけ", rows, 2)
if rows is not None:
rows = stage("5より大きいものだけ", rows, 5)
if rows is not None:
print("完了")
else:
print("停止")
# least を 2 に変えて、もう一度実行してください。
さらに: 確かめを足す
steps/pipeline.py に、天気の語が cloudy、rainy、sunny のどれでもないときに止まる段階を足します。table["weather"].isin(["cloudy", "rainy", "sunny"]).all() で比べます。
掃除したあとのデータで試し、次に strip の行の前で試します。この確かめは、行数が見逃したどんなことを見つけますか。