วิธีดึงข้อมูลปริมาณมากผ่านพร็อกซีแบบไม่ต้องเริ่มใหม่เมื่อการเชื่อมต่อขาด
บทความ
- บทนำ: ทำไมการดึงข้อมูลยาว ๆ ถึงมักขาดการเชื่อมต่อ และนั่นเป็นเรื่องปกติ
- การเตรียมการล่วงหน้า: เครื่องมือ สิทธิ์เข้าถึง และสภาพแวดล้อม
- แนวคิดพื้นฐาน: อภิธานศัพท์การดึงข้อมูลอย่างมั่นคงในภาษาง่าย ๆ
- ขั้นตอนที่ 1: ดาวน์โหลดต่อผ่าน http ด้วย header range
- ขั้นตอนที่ 2: เช็คพอยต์สำหรับการดึงข้อมูลแบบแบ่งหน้า
- ขั้นตอนที่ 3: idempotency เพื่อไม่ให้การรันซ้ำสร้างข้อมูลซ้ำ
- ขั้นตอนที่ 4: ลบข้อมูลซ้ำในผลลัพธ์โดยไม่ทำให้หน่วยความจำบวม
- ขั้นตอนที่ 5: การทำงานแบบขนานโดยไม่สูญเสียข้อมูล
- ขั้นตอนที่ 6: กลับมาทำต่อหลังหยุดไปนาน
- ขั้นตอนที่ 7: โครงสร้างตัวโหลดข้อมูลที่มั่นคงพร้อมใช้งานใน python
- ตรวจสอบผลลัพธ์: เช็คลิสต์การดึงข้อมูลที่มั่นคง
- ข้อผิดพลาดทั่วไปและวิธีแก้
- ความสามารถเพิ่มเติมและการปรับแต่ง
- Faq: คำถามที่พบบ่อยเกี่ยวกับการดึงข้อมูลที่มั่นคง
- บทสรุป: สิ่งที่คุณทำได้แล้วและก้าวต่อไป
บทนำ: ทำไมการดึงข้อมูลยาว ๆ ถึงมักขาดการเชื่อมต่อ และนั่นเป็นเรื่องปกติ
ถ้าคุณเคยรันการดึงข้อมูลขนาดใหญ่ คุณคงรู้สึกแบบนี้ กระบวนการดำเนินมาหลายชั่วโมง มาถึงเก้าสิบเปอร์เซ็นต์แล้วก็ขาดการเชื่อมต่อ การเชื่อมต่อล่ม เซิร์ฟเวอร์คืนข้อผิดพลาด แล็ปท็อปเข้าสู่โหมดสลีป และทุกอย่างต้องเริ่มใหม่หมด คู่มือนี้เขียนขึ้นเพื่อไม่ให้เรื่องแบบนี้เกิดขึ้นอีก
สิ่งที่คุณจะได้ในตอนท้าย คุณจะเรียนรู้วิธีสร้างตัวโหลดข้อมูลที่ทนทานต่อการขาดการเชื่อมต่อ มันดาวน์โหลดไฟล์ต่อจากตรงกลางได้ จำได้ว่าหยุดอยู่ที่หน้าไหน ไม่สร้างข้อมูลซ้ำเมื่อรันซ้ำ และสามารถกลับมาทำต่อได้แม้หลังจากหยุดไปนาน คุณจะได้โครงสร้าง Python ที่พร้อมใช้งาน ซึ่งสามารถปรับให้เข้ากับงานของคุณได้
คู่มือนี้เหมาะกับใคร สำหรับวิศวกร นักวิเคราะห์ และนักพัฒนาที่ดึงข้อมูลจาก API ดาวน์โหลดไฟล์ขนาดใหญ่ หรือรวบรวมผลลัพธ์แบบแบ่งหน้าผ่านพร็อกซี ระดับปานกลาง คุณควรเข้าใจพื้นฐาน HTTP และอ่านโค้ด Python ได้ ไม่จำเป็นต้องมีความรู้เชิงลึกด้านการเขียนโปรแกรมเครือข่าย
สิ่งที่ต้องรู้ล่วงหน้า Python พื้นฐาน แนวคิดเรื่อง HTTP request และ response หัวข้อ header และ status code คืออะไร ถ้าคุณเคยใช้ไลบรารี requests ก็เพียงพอแล้ว
ใช้เวลานานเท่าไหร่ อ่านและทำความเข้าใจแนวคิดประมาณสี่สิบนาที ประกอบตัวโหลดข้อมูลที่ใช้งานได้จากโครงสร้างของเราใช้เวลาหนึ่งถึงสามชั่วโมง ขึ้นอยู่กับแหล่งข้อมูลของคุณ
หมายเหตุสำคัญเกี่ยวกับหัวข้อ เราจะไม่พูดถึง status code อย่าง 429 และกลยุทธ์การลองใหม่แบบหน่วงเวลา (backoff) เรื่องนี้มีเนื้อหาแยกต่างหาก ที่นี่โฟกัสเรื่องเดียวเท่านั้น: สถานะของกระบวนการและการกลับมาทำต่อ วิธีบันทึกความคืบหน้า วิธีไม่ทำข้อมูลหายและไม่สร้างข้อมูลซ้ำ และวิธีทำต่อจากจุดที่หยุดไว้
เคล็ดลับ: เก็บสมุดโน้ตหรือไฟล์แยกไว้ใกล้มือ สำหรับจดพารามิเตอร์ของแหล่งข้อมูลคุณ เช่น รองรับการดาวน์โหลดต่อหรือไม่ มีการแบ่งหน้าหรือไม่ รูปแบบเคอร์เซอร์เป็นอย่างไร บันทึกเหล่านี้จะมีประโยชน์ในทุกขั้นตอน
การเตรียมการล่วงหน้า: เครื่องมือ สิทธิ์เข้าถึง และสภาพแวดล้อม
ก่อนเขียนโค้ด มาจัดเตรียมสภาพแวดล้อมการทำงานกันก่อน ใช้เวลาสิบนาที แต่จะประหยัดเวลาในการดีบักได้หลายชั่วโมง
สิ่งที่ต้องติดตั้ง
- ติดตั้ง Python เวอร์ชัน 3.10 ขึ้นไป ตรวจสอบเวอร์ชันด้วยคำสั่ง
python --versionในเทอร์มินัล - สร้าง virtual environment ด้วยคำสั่ง
python -m venv venvเพื่อไม่ให้ dependencies ของโปรเจกต์ปนกับของระบบ - เปิดใช้งาน environment บน Windows คือ
venv\Scripts\activateบน macOS และ Linux คือsource venv/bin/activate - ติดตั้งไลบรารีสำหรับ HTTP request ด้วยคำสั่ง
pip install requests - สำหรับการทำงานกับฐานข้อมูลสถานะที่เร็วขึ้น ไม่ต้องติดตั้งอะไรเพิ่ม เพราะโมดูล
sqlite3มีอยู่ในไลบรารีมาตรฐานของ Python แล้ว
สิ่งที่ต้องมีในเรื่องสิทธิ์เข้าถึง
- สิทธิ์เข้าถึงแหล่งข้อมูลของคุณ: URL, โทเคน หรือ API key ถ้าจำเป็น
- พร็อกซีจาก Proxeon พร้อมที่อยู่ พอร์ต และข้อมูลสำหรับยืนยันตัวตน หากไม่มีพร็อกซีที่เสถียร การดึงข้อมูลอย่างมั่นคงก็หมดความหมาย เพราะพร็อกซีคือสิ่งที่กระจายภาระและทำให้การเชื่อมต่อคาดเดาได้
- พื้นที่ดิสก์สำหรับไฟล์สถานะและสำหรับข้อมูลที่ดึงมา
การตรวจสอบพร็อกซี Proxeon
- นำสตริงการเชื่อมต่อรูปแบบ
http://логин:пароль@адрес:портมาใช้ - ทดสอบด้วยคำของ่าย ๆ ในเทอร์มินัลรัน
curl -x http://логин:пароль@адрес:порт https://api.ipify.orgและตรวจสอบว่าได้ IP ของพร็อกซีกลับมา ไม่ใช่ IP ของคุณเอง
เคล็ดลับ: เก็บสตริงการเชื่อมต่อพร็อกซีไว้ใน environment variable ไม่ใช่ในโค้ด วิธีนี้คุณจะไม่เผลอส่งรหัสผ่านเข้าไปในระบบควบคุมเวอร์ชัน ในโค้ดให้อ่านผ่าน os.environ
⚠️ ข้อควรระวัง: ทำงานเฉพาะกับแหล่งข้อมูลที่คุณมีสิทธิ์เข้าถึงตามกฎหมายเท่านั้น ปฏิบัติตามเงื่อนไขการใช้งานและขีดจำกัดที่เจ้าของบริการกำหนด พร็อกซี Proxeon ออกแบบมาสำหรับงานวิศวกรรมที่ถูกกฎหมาย: การกระจายภาระ ความเสถียรของการเชื่อมต่อ และการดึงข้อมูลอย่างถูกต้อง
✅ ตรวจสอบ: สภาพแวดล้อมพร้อมใช้งานถ้าคำสั่ง python -c "import requests, sqlite3" ทำงานโดยไม่มีข้อผิดพลาด และคำขอผ่านพร็อกซีคืนที่อยู่ของพร็อกซีเซิร์ฟเวอร์
แนวคิดพื้นฐาน: อภิธานศัพท์การดึงข้อมูลอย่างมั่นคงในภาษาง่าย ๆ
ก่อนเขียนโค้ด มาทำความเข้าใจคำศัพท์สำคัญกันก่อน ถ้าไม่มีสิ่งเหล่านี้ ขั้นตอนต่อไปจะฟังดูเหมือนคาถา
การดาวน์โหลดต่อ
การดาวน์โหลดต่อ คือการดาวน์โหลดไฟล์ต่อจากไบต์ที่มันหยุดไป แทนที่จะดาวน์โหลดไฟล์ใหม่หมด คุณขอให้เซิร์ฟเวอร์ส่งเฉพาะส่วนที่ขาด ทำงานผ่าน HTTP header ชื่อ Range
เช็คพอยต์
เช็คพอยต์ คือจุดความคืบหน้าที่บันทึกไว้ ลองนึกถึงการเซฟเกมคอมพิวเตอร์ ถ้ามีอะไรผิดพลาด คุณกลับไปที่เซฟล่าสุด ไม่ใช่เริ่มเกมใหม่ ในการดึงข้อมูล เช็คพอยต์เก็บว่าคุณหยุดอยู่ที่หน้าหรือเรกคอร์ดไหน
เคอร์เซอร์
เคอร์เซอร์ คือป้ายที่ API มอบให้คุณเพื่อให้คุณขอชุดข้อมูลถัดไปได้ มักเป็นสตริงอย่าง eyJvZmZzZXQiOjEwMH0 คุณส่งมันกลับไป แล้วเซิร์ฟเวอร์จะรู้ว่าจะทำต่อจากตรงไหน
Idempotency
Idempotency คือคุณสมบัติของการดำเนินการที่การทำซ้ำไม่เปลี่ยนผลลัพธ์ ถ้าคุณเขียนแถวเดิมด้วยคีย์เดิมสองครั้ง สุดท้ายจะมีแถวเดียว ไม่ใช่สองแถว นี่คือการป้องกันข้อมูลซ้ำเมื่อรันซ้ำ
การลบข้อมูลซ้ำ
การลบข้อมูลซ้ำ คือการกรองระเบียนที่ซ้ำกันออก แม้ทำงานอย่างระมัดระวัง วัตถุเดียวกันอาจมาได้สองครั้ง การลบข้อมูลซ้ำรับประกันว่าในชุดข้อมูลสุดท้ายจะเหลือเพียงหนึ่งเดียว
หลักการสำคัญ
ตัวโหลดข้อมูลที่มั่นคงสร้างบนแนวคิดเดียว: ต้องบันทึกความคืบหน้าอย่างต่อเนื่อง ไม่ใช่แค่ตอนจบ ทุกขั้นตอนอาจเป็นขั้นสุดท้ายก่อนการขาดการเชื่อมต่อ ดังนั้นหลังจากแต่ละส่วนที่สำเร็จ สถานะต้องถูกเขียนลงดิสก์ แล้วการกลับมาทำต่อก็แค่อ่านสถานะและดำเนินต่อ
เคล็ดลับ: จำกฎสามคำถามสำหรับการดึงข้อมูลทุกครั้ง ข้อแรก: ฉันหยุดที่ไหน? ข้อสอง: ทำอย่างไรไม่ให้ข้อมูลที่ได้มาแล้วซ้ำ? ข้อสาม: อะไรจะหมดอายุระหว่างที่ฉันไม่อยู่? คำตอบของคำถามเหล่านี้คือความมั่นคงนั่นเอง
ขั้นตอนที่ 1: ดาวน์โหลดต่อผ่าน HTTP ด้วย header Range
เป้าหมายของขั้นนี้ เรียนรู้ที่จะดาวน์โหลดไฟล์ขนาดใหญ่ให้ทำต่อจากไบต์ที่ยังไม่เสร็จหลังการขาดการเชื่อมต่อ ไม่ใช่เริ่มจากศูนย์
มันทำงานอย่างไร
HTTP อนุญาตให้ขอไม่ใช่ทั้งไฟล์ แต่ขอแค่บางส่วน สำหรับสิ่งนี้ header ชื่อ Range ถูกเพิ่มเข้าไปในคำขอ ตัวอย่างเช่น Range: bytes=1048576- หมายความว่า: ส่งทุกอย่างให้ฉันตั้งแต่ไบต์ที่ 1048576 เป็นต้นไป แต่ก่อนอื่นต้องตรวจสอบว่าเซิร์ฟเวอร์ทำได้
- ส่งคำขอเมธอด HEAD หรือ GET ปกติไปยังไฟล์ แล้วดู header ของการตอบกลับ
- หา header ชื่อ
Accept-Rangesถ้าค่าเป็นbytesแสดงว่าเซิร์ฟเวอร์รองรับการดาวน์โหลดต่อ - ถ้าไม่มี header นี้หรือค่าเป็น
noneการดาวน์โหลดต่อทำไม่ได้ ในกรณีนี้ต้องดาวน์โหลดทั้งไฟล์ในครั้งเดียวหรือหาแหล่งข้อมูลอื่น
ตรวจสอบการรองรับการดาวน์โหลดต่อ
นี่คือโค้ดที่ตรวจสอบว่าเซิร์ฟเวอร์ส่งบางส่วนของไฟล์ได้หรือไม่
import requests
def supports_resume(url, proxies):
resp = requests.head(url, proxies=proxies, timeout=30, allow_redirects=True)
accept = resp.headers.get("Accept-Ranges", "none")
total = resp.headers.get("Content-Length")
return accept.lower() == "bytes", totalดาวน์โหลดไฟล์ต่อจากตรงกลาง
ตอนนี้คือโค้ดหลัก มันดูว่าเราโหลดไปแล้วกี่ไบต์ในเครื่อง แล้วขอเฉพาะส่วนที่เหลือจากเซิร์ฟเวอร์
import os
import requests
def download_resumable(url, dest, proxies):
already = 0
if os.path.exists(dest):
already = os.path.getsize(dest)
headers = {}
if already > 0:
headers["Range"] = f"bytes={already}-"
mode = "ab" if already > 0 else "wb"
with requests.get(url, headers=headers, proxies=proxies,
stream=True, timeout=60) as r:
if already > 0 and r.status_code == 200:
mode = "wb"
already = 0
with open(dest, mode) as f:
for chunk in r.iter_content(chunk_size=65536):
if chunk:
f.write(chunk)
return os.path.getsize(dest)มาดูจุดสำคัญกัน ถ้าเซิร์ฟเวอร์คืนสถานะ 206 แสดงว่ามันส่งบางส่วนของไฟล์อย่างซื่อสัตย์และการเขียนต่อจะถูกต้อง ถ้าเซิร์ฟเวอร์คืน 200 แม้มี header Range แสดงว่ามันเพิกเฉยต่อการดาวน์โหลดต่อและส่งไฟล์ทั้งหมด ในกรณีนี้เราเปลี่ยนไปใช้โหมดเขียนทับทั้งหมด เพื่อไม่ให้ส่วนเก่าติดกับส่วนใหม่และทำให้ไฟล์เสียหาย
⚠️ ข้อควรระวัง: อย่าเขียนต่อในโหมด ab ถ้าคุณไม่แน่ใจว่าเซิร์ฟเวอร์ตอบด้วยรหัส 206 มิฉะนั้นคุณจะได้ไฟล์ที่เสียหาย ซึ่งส่วนต้นเป็นส่วนที่เหลือจากความพยายามครั้งก่อน และส่วนต่อมาเป็นไฟล์เต็มใหม่ ไฟล์แบบนี้จะเปิดไม่ได้ และคุณจะเสียเวลาในการหาสาเหตุ
เคล็ดลับ: ดาวน์โหลดไม่ใช่ตรงไปยังไฟล์เป้าหมาย แต่ไปยังไฟล์ชั่วคราวที่มีนามสกุล .part เมื่อดาวน์โหลดเสร็จสมบูรณ์แล้ว ให้เปลี่ยนชื่อเป็นชื่อสุดท้าย วิธีนี้คุณจะไม่มีทางสับสนระหว่างไฟล์ที่เสร็จแล้วกับไฟล์ที่ยังไม่เสร็จ
การตรวจสอบความสมบูรณ์
หลังดาวน์โหลดเสร็จ ควรตรวจสอบว่าไฟล์ไม่เสียหาย ถ้าเซิร์ฟเวอร์ส่ง header Content-Length ให้เปรียบเทียบกับขนาดจริงของไฟล์บนดิสก์ ถ้าขนาดตรงกัน ไฟล์มาครบถ้วน
def verify_size(dest, expected):
if expected is None:
return True
return os.path.getsize(dest) == int(expected)✅ ตรวจสอบ: หยุดการดาวน์โหลดกลางทางโดยปิดโปรแกรม เริ่มมันอีกครั้ง ในล็อกคุณควรเห็นว่าคำขอส่งด้วย header Range และไฟล์เขียนต่อ ไม่ได้เริ่มใหม่ ขนาดสุดท้ายตรงกับที่คาดไว้
ขั้นตอนที่ 2: เช็คพอยต์สำหรับการดึงข้อมูลแบบแบ่งหน้า
เป้าหมายของขั้นนี้ ตั้งค่าการบันทึกความคืบหน้าสำหรับ API ที่ส่งข้อมูลเป็นหน้า เพื่อให้ทำต่อจากหน้าที่ถูกต้องหลังการขาดการเชื่อมต่อ
สิ่งที่ต้องบันทึก
ไฟล์ดาวน์โหลดต่อตามไบต์ แต่การดึงข้อมูลแบบแบ่งหน้าใช้ตรรกะที่ต่างกันโดยสิ้นเชิง ที่นี่ไม่มีไบต์ มีหน้าและเรกคอร์ด ดังนั้นในเช็คพอยต์ต้องเก็บอย่างอื่น
- เคอร์เซอร์ ถ้า API ทำงานบนเคอร์เซอร์ นี่คือตัวเลือกที่เชื่อถือได้ที่สุด เพราะเคอร์เซอร์รู้เองว่าจะทำต่อจากไหน
- หมายเลขหน้าหรือ offset ถ้า API ทำงานบน offset และ limit ให้เก็บหมายเลขหน้าสุดท้ายที่ประมวลผลสำเร็จ
- ID ของเรกคอร์ดสุดท้าย ถ้าสามารถเรียงตาม ID หรือวันที่ที่เพิ่มขึ้นได้ คำขอถัดไปจะขอเรกคอร์ดที่มี ID มากกว่าที่บันทึกไว้
- ตัวนับเรกคอร์ดที่ประมวลผลแล้ว สำหรับการควบคุมและรายงาน
จะเก็บสถานะที่ไหน
คุณมีสามตัวเลือกหลัก จากง่ายไปมั่นคง
- ไฟล์ JSON ง่ายที่สุด เขียนดิกชันนารีที่มีเคอร์เซอร์และตัวนับลงไฟล์หลังแต่ละหน้า เหมาะสำหรับการดึงข้อมูลเดี่ยว ไม่ขนาน
- ฐานข้อมูล SQLite มั่นคงกว่า ให้ทรานแซกชัน ดังนั้นสถานะจะไม่เสียหายเมื่อขาดการเชื่อมต่อขณะเขียน เหมาะเมื่อมีข้อมูลมากและต้องการลบข้อมูลซ้ำ
- ฐานข้อมูลภายนอก สำหรับการดึงข้อมูลขนาดใหญ่แบบกระจาย เมื่อหลายโปรเซสแบ่งงานกัน
บันทึกเช็คพอยต์ใน JSON
import json
import os
def save_checkpoint(path, cursor, page, last_id, count):
tmp = path + ".tmp"
data = {
"cursor": cursor,
"page": page,
"last_id": last_id,
"count": count,
}
with open(tmp, "w") as f:
json.dump(data, f)
os.replace(tmp, path)
def load_checkpoint(path):
if not os.path.exists(path):
return {"cursor": None, "page": 0, "last_id": None, "count": 0}
with open(path) as f:
return json.load(f)สังเกตเทคนิคกับไฟล์ชั่วคราว เราเขียนลงไฟล์ที่มีนามสกุล .tmp แล้วเปลี่ยนชื่อแบบ atomic ผ่าน os.replace วิธีนี้ป้องกันสถานการณ์ที่โปรแกรมล่มขณะเขียนเช็คพอยต์ เช็คพอยต์เก่าจะยังคงสมบูรณ์ ไม่กลายเป็น JSON ครึ่ง ๆ กลาง ๆ ที่อ่านไม่ได้
⚠️ ข้อควรระวัง: อย่าเขียนเช็คพอยต์ตรงลงในไฟล์เดิมทับของเก่าโดยไม่มีไฟล์ชั่วคราว การขาดการเชื่อมต่อกลางการเขียนจะทิ้งเช็คพอยต์ที่เสียหายไว้ให้คุณ และการกลับมาทำต่อจะทำไม่ได้ การแทนที่แบบ atomic แก้ปัญหานี้ได้ทั้งหมด
เคล็ดลับ: บันทึกเช็คพอยต์เฉพาะหลังจากข้อมูลของหน้าเขียนลงที่เก็บจริงแล้ว ลำดับคือ: ได้หน้า เขียนข้อมูล แล้วค่อยอัปเดตเช็คพอยต์ ถ้าสลับลำดับ เมื่อขาดการเชื่อมต่อคุณจะข้ามหน้าและทำข้อมูลหาย
ลูปหลักพร้อมเช็คพอยต์
def paginate(fetch_page, save_data, cp_path, proxies):
cp = load_checkpoint(cp_path)
cursor = cp["cursor"]
count = cp["count"]
while True:
items, next_cursor = fetch_page(cursor, proxies)
if not items:
break
save_data(items)
count += len(items)
last_id = items[-1].get("id")
save_checkpoint(cp_path, next_cursor, cp["page"] + 1,
last_id, count)
cursor = next_cursor
if next_cursor is None:
break
return count✅ ตรวจสอบ: เริ่มการดึงข้อมูล ให้มันประมวลผลหลายหน้า แล้วหยุด เปิดไฟล์เช็คพอยต์และตรวจสอบว่ามีเคอร์เซอร์และตัวนับที่เป็นปัจจุบัน เริ่มอีกครั้ง: การดึงข้อมูลควรทำต่อจากเคอร์เซอร์ที่บันทึกไว้ ไม่ใช่จากหน้าแรก
ขั้นตอนที่ 3: Idempotency เพื่อไม่ให้การรันซ้ำสร้างข้อมูลซ้ำ
เป้าหมายของขั้นนี้ ทำให้การรันซ้ำหรือการทำคำขอเดิมซ้ำไม่นำไปสู่การเกิดระเบียนเดียวกันในที่เก็บของคุณ
ทำไมข้อมูลซ้ำจึงเกิดขึ้น
ลองนึกภาพ: คุณได้หน้าข้อมูล เขียนมันลงไฟล์ แต่โปรแกรมล่มก่อนที่เช็คพอยต์จะอัปเดต ในการรันครั้งถัดไปคุณจะขอหน้าเดิมอีกครั้ง ข้อมูลจะมาซ้ำและถูกเขียนซ้ำ นี่คือที่มาของข้อมูลซ้ำ มันเป็นผลสืบเนื่องที่หลีกเลี่ยงไม่ได้ของการขาดการเชื่อมต่อ และต้องจัดการในระดับสถาปัตยกรรม
คีย์สำหรับลบข้อมูลซ้ำ
เครื่องมือหลักของ idempotency คือ คีย์สำหรับลบข้อมูลซ้ำ นี่คือฟิลด์หรือชุดฟิลด์ที่ระบุระเบียนได้อย่างชัดเจน การเลือกคีย์ที่ถูกต้องแก้ปัญหาได้ครึ่งหนึ่ง
- ID ตามธรรมชาติ ถ้าระเบียนมีตัวระบุเฉพาะจากแหล่งข้อมูล ให้ใช้มัน นี่คือคีย์ที่สมบูรณ์แบบ
- ชุดฟิลด์ ถ้าไม่มี ID เดียว ให้สร้างคีย์จากหลายฟิลด์ที่เสถียร ตัวอย่างเช่น email บวกวันที่ลงทะเบียน
- แฮชของเนื้อหา ถ้าไม่มีฟิลด์ที่เสถียรเลย ให้คำนวณแฮชจากทั้งระเบียน นี่คือตัวเลือกสุดท้าย เพราะการเปลี่ยนแปลงฟิลด์ใดก็ตามจะสร้างคีย์ใหม่
เขียนโดยไม่มีข้อมูลซ้ำผ่าน UPSERT
ถ้าคุณเก็บผลลัพธ์ใน SQLite หรือฐานข้อมูลอื่น ให้ใช้การแทรกแบบเพิกเฉยต่อความขัดแย้ง แล้วการเขียนซ้ำด้วยคีย์เดิมจะไม่ทำอะไรเลย
import sqlite3
def init_db(path):
conn = sqlite3.connect(path)
conn.execute(
"CREATE TABLE IF NOT EXISTS records ("
"dedup_key TEXT PRIMARY KEY, payload TEXT)"
)
conn.commit()
return conn
def save_records(conn, items):
rows = [(item["id"], json.dumps(item)) for item in items]
conn.executemany(
"INSERT OR IGNORE INTO records (dedup_key, payload) "
"VALUES (?, ?)", rows
)
conn.commit()รายละเอียดสำคัญตรงนี้คือ PRIMARY KEY บนฟิลด์ dedup_key ฐานข้อมูลจะปฏิเสธการแทรกซ้ำด้วยคีย์เดิมเอง เพราะ INSERT OR IGNORE จะกลืนความขัดแย้งอย่างเงียบ ๆ คุณไม่ต้องตรวจสอบด้วยตนเองว่ามีระเบียนนี้อยู่แล้วหรือไม่ ฐานข้อมูลทำให้คุณและทำได้เร็ว
เคล็ดลับ: เลือกคีย์สำหรับลบข้อมูลซ้ำเพียงครั้งเดียวตอนเริ่มโปรเจกต์และบันทึกไว้ในเอกสาร การเปลี่ยนคีย์กลางการดึงข้อมูลหมายความว่าระเบียนเก่าและใหม่จะจับคู่กันไม่ได้ และข้อมูลซ้ำก็จะปรากฏขึ้น ความเสถียรของคีย์สำคัญกว่าความสวยงาม
✅ ตรวจสอบ: รันการดึงข้อมูลสองครั้งติดกันบนช่วงข้อมูลเดียวกัน นับจำนวนแถวในฐานข้อมูลด้วยคำสั่ง SELECT COUNT(*) FROM records ตัวเลขต้องเท่ากันหลังการรันครั้งแรกและครั้งที่สอง
ขั้นตอนที่ 4: ลบข้อมูลซ้ำในผลลัพธ์โดยไม่ทำให้หน่วยความจำบวม
เป้าหมายของขั้นนี้ กรองระเบียนที่ซ้ำกันบนข้อมูลหลายล้านแถว โดยไม่โหลดคีย์ที่เคยเห็นทั้งหมดเข้าสู่หน่วยความจำของคอมพิวเตอร์
วิธี naive และปัญหาของมัน
การลบข้อมูลซ้ำแบบง่ายที่สุด: เก็บเซ็ต set ของคีย์ทั้งหมดที่เคยเห็นไว้ในหน่วยความจำ สำหรับแต่ละระเบียนใหม่ ให้ตรวจสอบว่าคีย์อยู่ในเซ็ตหรือไม่ ทำงานได้ดีกับหลายแสนแถว แต่กับหลายล้านและหลายสิบล้านแถว เซ็ตจะขยายตัวและกิน RAM หลายกิกะไบต์ โปรแกรมจะช้าลงหรือล่ม
วิธีแรก: พึ่งพาฐานข้อมูล
วิธีที่ง่ายและมั่นคงที่สุดสำหรับปริมาณมาก คือไม่เก็บสิ่งที่เคยเห็นในหน่วยความจำเลย แต่ให้ฐานข้อมูลตรวจสอบผ่าน PRIMARY KEY อย่างที่เราทำในขั้นที่แล้ว ฐานข้อมูลเก็บดัชนีบนดิสก์ ไม่ใช่ในหน่วยความจำของโปรเซสคุณ มันจัดการคีย์หลายสิบล้านรายการได้โดยไม่เพิ่มภาระให้ RAM
วิธีที่สอง: แฮชของระเบียน
เมื่อไม่มีคีย์ตามธรรมชาติ ให้คำนวณแฮชขนาดกะทัดรัดจากระเบียน แฮชใช้พื้นที่คงที่และน้อยโดยไม่ขึ้นกับขนาดของระเบียน
import hashlib
import json
def record_hash(item):
raw = json.dumps(item, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()พารามิเตอร์ sort_keys=True ตรงนี้สำคัญมาก มันรับประกันว่าระเบียนที่มีเนื้อหาเดียวกันจะให้แฮชเดียวกัน แม้ฟิลด์ในนั้นจะเรียงลำดับต่างกัน ถ้าไม่มีการเรียงนี้ วัตถุสองอันที่เหมือนกันอาจได้แฮชต่างกันและผ่านไปเป็นระเบียนที่ต่างกัน
วิธีที่สาม: Bloom filter เพื่อประหยัดหน่วยความจำ
ถ้าคุณยังต้องการตรวจสอบที่รวดเร็วในหน่วยความจำบนปริมาณมหาศาล ใช้ Bloom filter นี่คือโครงสร้างที่ใช้พื้นที่น้อยและตอบเร็วว่าคุณเคยเห็นคีย์หรือไม่เคยเห็นแน่นอน มันมีคุณสมบัติพิเศษ: บางครั้งอาจบอกผิดว่าคีย์มีอยู่แล้วทั้งที่ไม่มี ดังนั้น Bloom filter จึงใช้เป็นการกรองเบื้องต้นอย่างรวดเร็ว ส่วนการตรวจสอบขั้นสุดท้ายให้ฐานข้อมูลทำ
- ตรวจสอบคีย์ด้วย Bloom filter
- ถ้าฟิลเตอร์บอกว่าไม่เคยเห็นแน่นอน ให้เขียนลงฐานข้อมูลทันที
- ถ้าฟิลเตอร์บอกว่าอาจเคยเห็น ให้ตรวจสอบอย่างแม่นยำในฐานข้อมูล
⚠️ ข้อควรระวัง: อย่าพยายามลบข้อมูลซ้ำหลายสิบล้านแถวด้วยเซ็ตธรรมดาในหน่วยความจำ บนแล็ปท็อปทั่วไปจะทำให้หน่วยความจำหมดและโปรเซสล่มกลางการดึงข้อมูล ให้ย้ายภาระไปที่ดิสก์ผ่านฐานข้อมูลหรือใช้ Bloom filter
เคล็ดลับ: ถ้าดึงข้อมูลเป็นชุดและภายในชุดเดียวอาจมีข้อมูลซ้ำ ให้ลบข้อมูลซ้ำในชุดด้วยเซ็ตธรรมดาในหน่วยความจำก่อนเขียนลงฐานข้อมูล ชุดไม่ใหญ่ หน่วยความจำจะไม่เสียหาย และการแทรกที่ไม่จำเป็นลงฐานข้อมูลจะน้อยลง
✅ ตรวจสอบ: รันการลบข้อมูลซ้ำบนชุดทดสอบขนาดใหญ่ที่มีข้อมูลซ้ำโดยเจตนา ตรวจสอบว่าจำนวนระเบียนที่ไม่ซ้ำสุดท้ายถูกต้อง และการใช้หน่วยความจำของโปรเซสคงที่ ไม่เพิ่มขึ้นเชิงเส้นตามจำนวนแถว
ขั้นตอนที่ 5: การทำงานแบบขนานโดยไม่สูญเสียข้อมูล
เป้าหมายของขั้นนี้ เร่งความเร็วการดึงข้อมูลด้วยคำขอแบบขนาน โดยไม่สูญเสียงานใด ๆ และลองงานที่ล้มเหลวใหม่อย่างถูกต้อง
คิวงาน
พื้นฐานของการทำงานแบบขนานอย่างปลอดภัยคือ คิวงาน คุณแบ่งงานออกเป็นชิ้นเล็ก ๆ ที่เป็นอิสระต่อกันล่วงหน้า ตัวอย่างเช่น รายการหน้า หรือช่วงข้อมูล ใส่ลงในคิว ผู้ปฏิบัติงานหลายตัวดึงงานจากคิว ทำเสร็จ แล้วเก็บผลลัพธ์ ถ้าผู้ปฏิบัติงานล่ม งานของมันสามารถคืนกลับเข้าคิวและมอบให้ตัวอื่นได้
จำกัดจำนวนพร้อมกัน
ไม่สามารถรันคำขอแบบขนานไม่จำกัดจำนวนได้ นั่นจะทำให้แหล่งข้อมูลและพร็อกซีของคุณโอเวอร์โหลด วิธีที่ถูกต้องคือจำกัดจำนวนผู้ปฏิบัติงานพร้อมกันให้เป็นค่าที่สมเหตุสมผล เริ่มจากจำนวนน้อยแล้วค่อย ๆ เพิ่มพร้อมสังเกตความเสถียร
from concurrent.futures import ThreadPoolExecutor, as_completed
def run_parallel(tasks, worker, proxies, max_workers=5):
results = []
failed = []
with ThreadPoolExecutor(max_workers=max_workers) as pool:
future_map = {
pool.submit(worker, t, proxies): t for t in tasks
}
for future in as_completed(future_map):
task = future_map[future]
try:
results.append(future.result())
except Exception:
failed.append(task)
return results, failedลองงานที่ล้มเหลวใหม่
รายการ failed ที่รวบรวมมาไม่ใช่ข้อมูลที่สูญหาย แต่เป็นรายการสิ่งที่ต้องลองใหม่ หลังรอบแรกคุณรันงานที่ล้มเหลวอีกครั้ง ปกติแล้วก็เพียงพอที่จะเก็บส่วนที่เหลือให้ครบ
def run_with_retry(tasks, worker, proxies, rounds=3):
remaining = tasks
for _ in range(rounds):
done, remaining = run_parallel(remaining, worker, proxies)
if not remaining:
break
return remainingบทบาทของพร็อกซี Proxeon ในการทำงานแบบขนาน เมื่อทำงานแบบขนาน พร็อกซีจะกระจายการเชื่อมต่อ ทำให้การดึงข้อมูลเสถียรและคาดเดาได้มากขึ้น ผู้ปฏิบัติงานแต่ละตัวทำงานผ่านการเชื่อมต่อของตัวเอง และภาระไม่กระจุกตัวที่จุดเดียว
⚠️ ข้อควรระวัง: เมื่อเขียนแบบขนานลงไฟล์เดียวหรือเช็คพอยต์เดียว จะเกิด race condition ผู้ปฏิบัติงานสองตัวอาจเขียนทับสถานะของกันและกัน ให้เขียนผลลัพธ์ลงฐานข้อมูลที่มีทรานแซกชันเท่านั้น หรือใช้ไฟล์แยกสำหรับผู้ปฏิบัติงานแต่ละตัว แล้วรวบรวมเช็คพอยต์รวมด้วยเธรดแยก
เคล็ดลับ: ทำให้งานเล็กและเป็นอิสระ ถ้างานหนึ่งครอบคลุมช่วงใหญ่เกินไป การขาดการเชื่อมต่อจะทิ้งงานไปมาก งานเล็ก ๆ ลองใหม่ได้ถูกและแทบไม่รู้สึก
✅ ตรวจสอบ: เริ่มการดึงข้อมูลแบบขนาน ทำให้ผู้ปฏิบัติงานบางตัวล่มโดยเจตนา หลังรอบลองใหม่ รายการ remaining ควรว่างเปล่า และชุดข้อมูลสุดท้ายครบถ้วน เปรียบเทียบจำนวนระเบียนที่ได้กับที่คาดไว้
ขั้นตอนที่ 6: กลับมาทำต่อหลังหยุดไปนาน
เป้าหมายของขั้นนี้ ทำการดึงข้อมูลต่ออย่างถูกต้อง ถ้าระหว่างความพยายามครั้งต่าง ๆ ผ่านไปนาน และเข้าใจว่าอะไรอาจหมดอายุในช่วงเวลานั้น
อะไรหมดอายุตามเวลา
การขาดการเชื่อมต่อหนึ่งนาทีกับหยุดพักหนึ่งวันเป็นสถานการณ์ที่ต่างกัน ในช่วงหยุดพักนาน ส่วนหนึ่งของสถานะคุณอาจใช้ไม่ได้อีกต่อไป
- เซสชัน บริการหลายแห่งเก็บเซสชันไว้ในเวลาจำกัด หลังหยุดนาน เซิร์ฟเวอร์จะลืมมัน และคำขอจะเริ่มคืนข้อผิดพลาดการยืนยันตัวตน
- โทเคนเข้าถึง โทเคน API มักมีอายุเป็นนาทีหรือชั่วโมง โทเคนที่หมดอายุต้องอัปเดตก่อนทำต่อ
- เคอร์เซอร์ เคอร์เซอร์บางตัวอายุสั้น ถ้าเคอร์เซอร์หมดอายุ ต้องเริ่มจากจุดที่เสถียรที่สุดที่ใกล้ที่สุด เช่น ID ของเรกคอร์ดสุดท้าย
- ตัวข้อมูลเอง ระหว่างหยุดพัก อาจมีระเบียนใหม่ปรากฏในแหล่งข้อมูลหรือระเบียนเก่าเปลี่ยนแปลง สิ่งนี้ส่งผลต่อ offset ในการนำทางแบบแบ่งหน้า
กลยุทธ์การกลับมาทำต่ออย่างปลอดภัย
- ตอนเริ่ม ตรวจสอบอายุของเช็คพอยต์ ถ้ามันเก่า เตรียมพร้อมว่าส่วนหนึ่งของสถานะหมดอายุแล้ว
- อัปเดตโทเคนเข้าถึงและสร้างเซสชันใหม่ก่อนคำขอแรก อย่าพึ่งพาของเก่า
- ให้ความสำคัญกับการกลับมาทำต่อด้วย ID ของเรกคอร์ดสุดท้าย ไม่ใช่หมายเลขหน้า ID เสถียร แต่หมายเลขหน้าเลื่อนเมื่อข้อมูลเปลี่ยน
- ทำคำขอทดสอบด้วยเคอร์เซอร์ที่บันทึกไว้ ถ้ามันคืนข้อผิดพลาดเคอร์เซอร์ไม่ถูกต้อง ให้เปลี่ยนไปกลับมาทำต่อด้วย last_id
def resume(cp, fetch_by_id, fetch_by_cursor, proxies):
if cp["cursor"]:
try:
return fetch_by_cursor(cp["cursor"], proxies)
except CursorExpired:
pass
return fetch_by_id(cp["last_id"], proxies)ทำไมการกลับมาทำต่อด้วย ID จึงเชื่อถือได้กว่า ลองนึกภาพว่าคุณหยุดที่หน้า 50 ในการเรียงตามวันที่ ระหว่างที่คุณไม่อยู่ มีระเบียนใหม่เพิ่มเข้ามาที่ต้น ตอนนี้หน้า 50 มีข้อมูลที่ต่างออกไป และคุณจะข้ามระเบียนบางส่วน การกลับมาทำต่อด้วย ID ของเรกคอร์ดสุดท้ายไม่ได้รับผลกระทบจากสิ่งนี้: คุณแค่ขอทุกอย่างที่มี ID มากกว่าที่บันทึกไว้
เคล็ดลับ: เก็บทั้งเคอร์เซอร์และ ID ของเรกคอร์ดสุดท้ายในเช็คพอยต์พร้อมกันเสมอ เคอร์เซอร์เร็วกว่า แต่ ID คือเชือกนิรภัยของคุณในกรณีที่เคอร์เซอร์หมดอายุระหว่างหยุดพักนาน
✅ ตรวจสอบ: หยุดการดึงข้อมูล รอให้นานพอที่โทเคนหรือเคอร์เซอร์จะหมดอายุ แล้วเริ่มใหม่ ตัวโหลดข้อมูลควรอัปเดตโทเคน ตรวจพบเคอร์เซอร์ที่หมดอายุ และทำต่อด้วย ID โดยไม่สูญเสียและไม่ซ้ำระเบียน
ขั้นตอนที่ 7: โครงสร้างตัวโหลดข้อมูลที่มั่นคงพร้อมใช้งานใน Python
เป้าหมายของขั้นนี้ รวบรวมทุกอย่างที่เรียนมาเป็นโครงที่ทำงานได้ ซึ่งคุณจะปรับให้เข้ากับแหล่งข้อมูลของคุณ
ด้านล่างคือโครงที่รวมเช็คพอยต์ การลบข้อมูลซ้ำผ่านฐานข้อมูล การอัปเดตโทเคน และการกลับมาทำต่อ ฟังก์ชันดึงหน้าคุณใส่ของคุณเองตาม API ที่ใช้
import os
import json
import sqlite3
import requests
class ResilientLoader:
def __init__(self, cp_path, db_path, proxies):
self.cp_path = cp_path
self.proxies = proxies
self.conn = sqlite3.connect(db_path)
self.conn.execute(
"CREATE TABLE IF NOT EXISTS records ("
"dedup_key TEXT PRIMARY KEY, payload TEXT)"
)
self.conn.commit()
def load_cp(self):
if not os.path.exists(self.cp_path):
return {"cursor": None, "last_id": None, "count": 0}
with open(self.cp_path) as f:
return json.load(f)
def save_cp(self, cp):
tmp = self.cp_path + ".tmp"
with open(tmp, "w") as f:
json.dump(cp, f)
os.replace(tmp, self.cp_path)
def save_records(self, items):
rows = [(str(i["id"]), json.dumps(i)) for i in items]
self.conn.executemany(
"INSERT OR IGNORE INTO records "
"(dedup_key, payload) VALUES (?, ?)", rows
)
self.conn.commit()
def run(self, fetch_page):
cp = self.load_cp()
while True:
items, next_cursor = fetch_page(
cp["cursor"], cp["last_id"], self.proxies
)
if not items:
break
self.save_records(items)
cp["count"] += len(items)
cp["last_id"] = items[-1]["id"]
cp["cursor"] = next_cursor
self.save_cp(cp)
if next_cursor is None:
break
return cp["count"]ตัวอย่างฟังก์ชันดึงหน้าสำหรับแหล่งข้อมูลของคุณ ที่นี่คุณใช้งานตรรกะคำขอและการแยกการตอบกลับ
def fetch_page(cursor, last_id, proxies):
params = {"limit": 100}
if cursor:
params["cursor"] = cursor
elif last_id:
params["after_id"] = last_id
r = requests.get(
"https://example-source/api/records",
params=params, proxies=proxies, timeout=60
)
r.raise_for_status()
data = r.json()
return data["items"], data.get("next_cursor")การเริ่มกลไกทั้งหมดดูง่ายมาก
proxies = {
"http": os.environ["PROXEON_URL"],
"https": os.environ["PROXEON_URL"],
}
loader = ResilientLoader("state.json", "out.db", proxies)
total = loader.run(fetch_page)
print("Всего записей:", total)เคล็ดลับ: เพิ่มการบันทึกทุก ๆ ร้อยระเบียนในลูป: เวลา ตัวนับ เคอร์เซอร์ปัจจุบัน วิธีนี้คุณจะเห็นความคืบหน้าและเข้าใจได้ง่าย ๆ ถ้าการดึงข้อมูลค้างอยู่ที่จุดเดียว
✅ ตรวจสอบ: เริ่มโครงบนแหล่งข้อมูลจริง หยุดกลางทาง เริ่มใหม่ จำนวนระเบียนสุดท้ายหลังดาวน์โหลดต่อจะตรงกับจำนวนระเบียนทั้งหมดของแหล่งข้อมูล และการรันซ้ำจะไม่เพิ่มตัวนับของแถวที่ไม่ซ้ำ
ตรวจสอบผลลัพธ์: เช็คลิสต์การดึงข้อมูลที่มั่นคง
ไล่ดูรายการนี้ ถ้าทุกข้อเป็นจริง ตัวโหลดข้อมูลของคุณมั่นคงจริง ๆ
- การดาวน์โหลดไฟล์ต่อทำต่อจากไบต์ที่ยังไม่เสร็จ ไม่ใช่จากศูนย์
- ตัวโหลดข้อมูลจัดการกรณีที่เซิร์ฟเวอร์เพิกเฉยต่อ header Range อย่างถูกต้อง
- เช็คพอยต์ถูกบันทึกหลังแต่ละหน้าที่ประมวลผล ไม่ใช่แค่ตอนจบ
- เช็คพอยต์เขียนแบบ atomic ผ่านไฟล์ชั่วคราวและการเปลี่ยนชื่อ
- การรันซ้ำไม่สร้างข้อมูลซ้ำในชุดสุดท้าย
- การลบข้อมูลซ้ำไม่เพิ่มหน่วยความจำเชิงเส้นตามจำนวนแถว
- ผู้ปฏิบัติงานแบบขนานไม่สูญเสียงานที่ล้มเหลวและลองใหม่
- หลังหยุดนาน โทเคนอัปเดต และเคอร์เซอร์ที่หมดอายุถูกแทนที่ด้วยการกลับมาทำต่อด้วย ID
วิธีทดสอบ
- รันการดึงข้อมูลชุดเล็กแบบเต็มและจำจำนวนระเบียน
- รันอีกครั้งบนชุดเดียวกันและตรวจสอบว่าจำนวนไม่เปลี่ยน
- หยุดการดึงข้อมูลที่จุดต่าง ๆ: ตอนเริ่ม กลางทาง ใกล้จบ
- หลังการหยุดแต่ละครั้ง ให้รีสตาร์ทและตรวจสอบว่าผลสุดท้ายเหมือนกัน
ตัวชี้วัดความสำเร็จ จำนวนระเบียนที่ไม่ซ้ำเสถียรระหว่างการรัน การใช้หน่วยความจำไม่เพิ่มขึ้นอย่างไม่ควบคุม การกลับมาทำต่อทำต่อจากจุดที่บันทึกไว้เสมอ ไม่มีไฟล์เสียหายหลังการดาวน์โหลดต่อ
ข้อผิดพลาดทั่วไปและวิธีแก้
ปัญหา: ไฟล์หลังดาวน์โหลดต่อเปิดไม่ได้ สาเหตุ: ข้อมูลถูกเขียนต่อในโหมด ab แม้เซิร์ฟเวอร์ตอบด้วยรหัส 200 และส่งไฟล์ทั้งหมด วิธีแก้: ตรวจสอบสถานะการตอบกลับ เมื่อเป็น 200 ให้เปลี่ยนไปเขียนทับไฟล์ทั้งหมดจากศูนย์
ปัญหา: หลังขาดการเชื่อมต่อ การดึงข้อมูลเริ่มจากหน้าแรก สาเหตุ: เช็คพอยต์ถูกบันทึกแค่ตอนจบหรือไม่ถูกบันทึกเลย วิธีแก้: บันทึกเช็คพอยต์หลังแต่ละหน้าที่ประมวลผล ทันทีหลังเขียนข้อมูล
ปัญหา: เช็คพอยต์อ่านไม่ได้ JSON เสีย สาเหตุ: โปรแกรมล่มขณะเขียนตรงลงในไฟล์เป้าหมาย วิธีแก้: เขียนลงไฟล์ชั่วคราวและแทนที่แบบ atomic ผ่าน os.replace
ปัญหา: มีข้อมูลซ้ำในชุดสุดท้าย สาเหตุ: ไม่มีคีย์สำหรับลบข้อมูลซ้ำหรือมันไม่เสถียร วิธีแก้: ตั้ง PRIMARY KEY บนคีย์ที่เชื่อถือได้และใช้ INSERT OR IGNORE
ปัญหา: โปรเซสล่มเพราะหน่วยความจำไม่พอบนปริมาณมาก สาเหตุ: คีย์ทั้งหมดที่เคยเห็นถูกเก็บในเซ็ตในหน่วยความจำ วิธีแก้: ย้ายการตรวจสอบความไม่ซ้ำไปที่ฐานข้อมูลหรือใช้ Bloom filter
ปัญหา: หลังหยุดนาน คำขอคืนข้อผิดพลาดการยืนยันตัวตน สาเหตุ: โทเคนหรือเซสชันหมดอายุระหว่างหยุดพัก วิธีแก้: อัปเดตโทเคนและสร้างเซสชันใหม่ทุกครั้งที่เริ่ม
ปัญหา: หลังหยุดพัก ระเบียนบางส่วนถูกข้ามหรือซ้ำ สาเหตุ: การกลับมาทำต่อใช้หมายเลขหน้า แต่ข้อมูลในแหล่งข้อมูลเปลี่ยนไป วิธีแก้: กลับมาทำต่อด้วย ID ของเรกคอร์ดสุดท้าย ไม่ใช่ offset
ปัญหา: ผู้ปฏิบัติงานแบบขนานสูญเสียข้อมูลบางส่วน สาเหตุ: ผู้ปฏิบัติงานหลายตัวเขียนเช็คพอยต์เดียวกันและเขียนทับกัน วิธีแก้: เขียนผลลัพธ์ลงฐานข้อมูลที่มีทรานแซกชัน ไม่ใช่ไฟล์สถานะรวม
ความสามารถเพิ่มเติมและการปรับแต่ง
การเขียนเป็นชุด
อย่าเขียนลงฐานข้อมูลทีละแถว รวบรวมชุดหลายร้อยระเบียนและแทรกพร้อมกันผ่าน executemany วิธีนี้ทำให้การเขียนเร็วขึ้นหลายเท่าบนปริมาณมาก
การ commit เป็นระยะ
เรียก commit ไม่ใช่ทุกการเขียน แต่ทุกสองสามร้อยแถว commit บ่อยเกินไปทำให้ฐานข้อมูลช้า ห่างเกินไปเสี่ยงสูญเสียข้อมูลมากขึ้นเมื่อขาดการเชื่อมต่อ หาสมดุลให้เหมาะกับภาระของคุณ
รายงานความคืบหน้า
เพิ่มการประเมินเวลาที่เหลือ เมื่อรู้ความเร็วในการประมวลผลหน้าและจำนวนระเบียนทั้งหมด คุณจะประมาณได้ว่าจะรออีกนานแค่ไหน สะดวกสำหรับการดึงข้อมูลยาว ๆ
แยกที่เก็บข้อมูลดิบและข้อมูลที่ประมวลผลแล้ว
เก็บการตอบกลับดิบแยกจากระเบียนที่แยกแล้ว ถ้าภายหลังคุณเปลี่ยนตรรกะการแยก คุณไม่ต้องดาวน์โหลดข้อมูลใหม่ แค่ประมวลผลการตอบกลับดิบผ่าน parser ใหม่ก็เพียงพอ
เคล็ดลับ: ตั้งค่าพร็อกซี Proxeon ให้การเชื่อมต่อเสถียรตลอดการดึงข้อมูล การเชื่อมต่อที่เสถียรลดจำนวนการขาดการเชื่อมต่อ หมายความว่าตัวโหลดข้อมูลของคุณเข้าสู่การกลับมาทำต่อน้อยลงและทำงานเร็วขึ้น
FAQ: คำถามที่พบบ่อยเกี่ยวกับการดึงข้อมูลที่มั่นคง
จะรู้ได้อย่างไรว่าเซิร์ฟเวอร์รองรับการดาวน์โหลดไฟล์ต่อ? ส่งคำขอ HEAD และดู header Accept-Ranges ค่า bytes หมายถึงรองรับ การไม่มี header หรือค่า none หมายความว่าการดาวน์โหลดต่อทำไม่ได้
ทำอย่างไรถ้า API ไม่ให้เคอร์เซอร์ ให้แค่หน้า? เก็บหมายเลขหน้าและถ้าเป็นไปได้ ID ของเรกคอร์ดสุดท้าย ให้กลับมาทำต่อด้วย ID เป็นหลัก เพราะหมายเลขหน้าเลื่อนเมื่อข้อมูลเปลี่ยน
ควรบันทึกเช็คพอยต์บ่อยแค่ไหน? หลังแต่ละหน้าที่ประมวลผลและเขียนสำเร็จ วิธีนี้เมื่อขาดการเชื่อมต่อคุณจะสูญเสียงานมากสุดหนึ่งหน้า ไม่ใช่ทั้งการดึงข้อมูล
ลบข้อมูลซ้ำโดยไม่มีฐานข้อมูลได้ไหม? บนปริมาณน้อยได้ ด้วยเซ็ตธรรมดาในหน่วยความจำ บนหลายล้านแถวอันตรายเพราะหน่วยความจำ ควรใช้ฐานข้อมูลที่มี PRIMARY KEY หรือ Bloom filter
จะเลือกอะไรเป็นคีย์สำหรับลบข้อมูลซ้ำ? ID เฉพาะตามธรรมชาติของแหล่งข้อมูลถ้ามี ถ้าไม่มี ให้ใช้ชุดฟิลด์ที่เสถียร ในกรณีสุดท้าย ใช้แฮชของทั้งระเบียนพร้อมเรียงคีย์
ทำไมการกลับมาทำต่อด้วย ID จึงเชื่อถือได้กว่าหมายเลขหน้า? เพราะข้อมูลในแหล่งข้อมูลอาจเปลี่ยน ระเบียนใหม่เลื่อนหน้า และตามหมายเลขคุณจะข้ามหรือซ้ำข้อมูล ID ไม่ขึ้นกับสิ่งนี้
ควรตั้งผู้ปฏิบัติงานแบบขนานกี่ตัว? เริ่มจากจำนวนน้อยและเพิ่มพร้อมสังเกตความเสถียรและขีดจำกัดของแหล่งข้อมูล ความขนานที่มากเกินไปทำร้ายมากกว่าช่วย
จะเก็บสตริงการเชื่อมต่อพร็อกซีอย่างปลอดภัยได้อย่างไร? ใน environment variable ไม่ใช่ในโค้ด อ่านผ่าน os.environ วิธีนี้รหัสผ่านจะไม่เข้าไปในระบบควบคุมเวอร์ชัน
ทำอย่างไรถ้าเคอร์เซอร์หมดอายุระหว่างหยุดพักนาน? จับข้อผิดพลาดเคอร์เซอร์ไม่ถูกต้องและเปลี่ยนไปกลับมาทำต่อด้วย ID ของเรกคอร์ดสุดท้ายที่บันทึกไว้
ต้องตรวจสอบความสมบูรณ์ของไฟล์ที่ดาวน์โหลดหรือไม่? ต้อง เปรียบเทียบขนาดจริงของไฟล์กับ header Content-Length ถ้าเซิร์ฟเวอร์ให้ checksum ให้ตรวจสอบด้วย
บทสรุป: สิ่งที่คุณทำได้แล้วและก้าวต่อไป
คุณเดินทางจาการดึงข้อมูลที่เปราะบาง ซึ่งพังเมื่อขาดการเชื่อมต่อครั้งแรก สู่ตัวโหลดข้อมูลที่มั่นคง ตอนนี้คุณมีเครื่องมือทั้งหมดเพื่อให้การขาดการเชื่อมต่อไม่ใช่หายนะอีกต่อไป แต่เป็นสถานการณ์ทำงานปกติ
สิ่งที่คุณเชี่ยวชาญแล้ว การดาวน์โหลดไฟล์ต่อผ่าน header Range พร้อมการตรวจสอบ Accept-Ranges การบันทึกความคืบหน้าผ่านเช็คพอยต์แบบ atomic Idempotency ที่ระดับคีย์สำหรับลบข้อมูลซ้ำ การลบข้อมูลซ้ำบนหลายล้านแถวโดยไม่ทำให้หน่วยความจำบวม การทำงานแบบขนานพร้อมลองงานที่ล้มเหลวใหม่ การกลับมาทำต่อหลังหยุดนานพร้อมอัปเดตโทเคนและเปลี่ยนเคอร์เซอร์ที่หมดอายุ และที่สำคัญที่สุด โครง Python พร้อมใช้งานที่รวมทั้งหมดนี้เข้าด้วยกัน
ทำอะไรต่อไป นำแหล่งข้อมูลจริงของคุณมาและปรับฟังก์ชันดึงหน้าให้เข้ากับมัน เริ่มจากปริมาณน้อย ดีบักการกลับมาทำต่อบนการหยุดชะงัก แล้วค่อยขยาย ตั้งค่าพร็อกซี Proxeon ที่เสถียรเพื่อให้การเชื่อมต่อคาดเดาได้ตลอดการดึงข้อมูล
จะพัฒนาต่อไปไหน ศึกษา การเขียนเป็นชุด และ Bloom filter ให้ลึกขึ้น เพิ่มการติดตามความคืบหน้าและประเมินเวลา แยกการเก็บข้อมูลดิบและข้อมูลที่ประมวลผลแล้ว ค่อย ๆ
