Scrapeless और DuckDB के साथ एक स्क्रैप किए गए डेटा एनालिटिक्स पाइपलाइन बनाएं।

TL;DR:

• एक स्क्रैपिंग पाइपलाइन जो JSON फ़ाइल में समाप्त होती है, केवल आधी पाइपलाइन होती है; विश्लेषणात्मक प्रश्न तब आते हैं जब डेटा सहेजा जाता है।
• डकडीबी सीधे JSON लाइनों को पढ़ता है, इसलिए स्क्रैप की गई रिकॉर्ड एक क्वेरी करने योग्य तालिका बन जाती है जिसमें कोई सर्वर, कोई स्कीमा माइग्रेशन, और कोई ETL टूल नहीं होता।
• यह पाइपलाइन 3 पृष्ठों को Scrapeless Universal Scraping API के माध्यम से प्राप्त करती है, 30 रिकॉर्ड निकालती है, उन्हें डकडीबी में लोड करती है, और पार्क्वेट में लिखती है।
• ZSTD संपीड़न के साथ पार्क्वेट ने 4,674 बाइट्स में 30 रिकॉर्ड संग्रहीत किए, जबकि JSON लाइनों के लिए 7,690 बाइट्स लगे, और डकडीबी फ़ाइल को सीधे क्वेरी करता है बिना उसे फिर से लोड किए।
• हर चरण नीचे आपके मशीन पर चलता है बिना किसी क्लाउड वेयरहाउस खाते और बिना किसी प्रमाण पत्र के सिवाय एक स्क्रैपलेस कुंजी के।
• स्क्रैपलेस फ्रीम योजना पर प्रारंभ करें और फ़ेच चरण को अपने खुद के स्रोत पर इंगित करें।

पाइपलाइन एक नज़र में

प्रवाह पाँच चरणों में है, और दिलचस्प डिजाइन निर्णय वह है जहाँ डेटा टेक्स्ट बनना बंद करता है और तालिका बनना शुरू होता है।

फेच (स्क्रापलेस) → रिकॉर्ड खोजें → फ़ील्ड निकालें → टाइप की गई तालिका (डकडीबी) में स्थानांतरित करें → पार्क्वेट के रूप में संग्रहित करें

JSON लाइन्स स्क्रैपिंग आधे और विश्लेषणात्मक आधे के बीच का हैंडऑफ़ फ़ॉर्मेट है। यह ऐप्पेंड-फ्रेंडली है, क्रैश रन का सामना करता है बिना पिछले डेटा को भ्रष्ट किए, और डकडीबी इसे मूल रूप से पढ़ता है - इसलिए लिखने के लिए कोई लोडर नहीं है। फ़ॉर्मेट JSON लाइन्स स्पेकिफिकेशन पर निर्दिष्ट है।

यहाँ कुछ भी वेयरहाउस खाते की आवश्यकता नहीं है। डकडीबी इन-प्रोसेस चलता है, जो स्क्रैप की गई डेटा पर वास्तविक SQL प्राप्त करने का सबसे सस्ता तरीका बनाता है। जब गंतव्य एक प्रबंधित वेयरहाउस होता है, तो स्नोफ्लेक इनजेशन गाइड समान संग्रहण चरण को एक अलग लैंडिंग क्षेत्र के साथ कवर करता है।

आवश्यकताएँ

• Python 3.9 या बाद का।
• डैशबोर्ड से एक स्क्रैपलेस API कुंजी।
• duckdb स्थापित करें:

pip install "duckdb==1.5.4"

कुंजी सेट करें:

export SCRAPELESS_API_KEY="आपकी_api_की_यहाँ"

चरण 1-3: फ़ेच, खोजें, निकालें

स्क्रैपलेस यूनिवर्सल स्क्रैपिंग API के माध्यम से हर पृष्ठ के लिए एक कॉल प्रदर्शित HTML लौटाता है, और एक मानक-लाइब्रेरी पार्सर प्रत्येक उद्धरण ब्लॉक को एक रिकॉर्ड में बदल देता है। एक्सट्रैक्टर प्रति पंक्ति एक JSON ऑब्जेक्ट निकालता है:

import json, os, urllib.request
from html.parser import HTMLParser

API = "https://api.scrapeless.com/api/v2/unlocker/request"

def fetch(url: str) -> str:
    payload = json.dumps({
        "actor": "unlocker.webunlocker",
        "input": {"url": url, "js_render": True, "headless": True},
    }).encode()
    req = urllib.request.Request(
        API, data=payload,
        headers={"x-api-token": os.environ["SCRAPELESS_API_KEY"],
                 "Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.loads(r.read())["data"]

class QuoteParser(HTMLParser):
    def __init__(self):
        super().__init__()
        self.rows, self._cur, self._cap = [], None, None
    def handle_starttag(self, tag, attrs):
        a = dict(attrs); cls = a.get("class", "")
        if tag == "div" and "quote" in cls:
            self._cur = {"text": "", "author": "", "tags": []}
        elif self._cur is not None and tag == "span" and "text" in cls:
            self._cap = "text"
        elif self._cur is not None and tag == "small" and "author" in cls:
            self._cap = "author"
        elif self._cur is not None and tag == "a" and "tag" in cls:
            self._cap = "tag"
    def handle_data(self, data):
        if self._cap == "text": self._cur["text"] += data
        elif self._cap == "author": self._cur["author"] += data
        elif self._cap == "tag": self._cur["tags"].append(data.strip())
    def handle_endtag(self, tag):
        if self._cap in ("text", "author", "tag"): self._cap = None
        if tag == "div" and self._cur and self._cur["text"]:
            self.rows.append(self._cur); self._cur = None

rows = []
for page in range(1, 4):
    p = QuoteParser(); p.feed(fetch(f"https://quotes.toscrape.com/page/{page}/"))
    rows.extend({"page": page, **r} for r in p.rows)

print(f"पृष्ठ प्राप्त: 3 | निकाले गए रिकॉर्ड: {len(rows)}")
with open("quotes.jsonl", "w", encoding="utf-8") as f:
    for r in rows: f.write(json.dumps(r, ensure_ascii=False) + "\n")
print(f"quotes.jsonl में लिखा गया ({os.path.getsize('quotes.jsonl')} बाइट्स)")

पृष्ठ प्राप्त: 3 | निकाले गए रिकॉर्ड: 30
quotes.jsonl में लिखा गया (7690 बाइट्स)

यहाँ दो विकल्प हैं जिन्हें अपने स्वयं के संस्करण में रखना उचित है।
पृष्ठ संख्या प्रत्येक रिकॉर्ड के साथ निकासी समय पर संलग्न होती है। उत्पत्ति संग्रह के बिंदु पर रिकॉर्ड करने के लिए लगभग मुफ्त है और बाद में पुनर्निर्माण करने के लिए महंगा है, और यह वही है जो आपको "यह किस पृष्ठ से आया" का उत्तर देने की अनुमति देता है बिना कुछ फिर से चलाए।

ensure_ascii=False टाइपोग्राफिक उद्धरण चिह्नों को सुरक्षित रखते हैं बजाय इसके कि उन्हें भागों में बाँट दें। जब पाठ डेटा हो तब यह महत्वपूर्ण है — भागों में सुरक्षित आउटपुट अभी भी मान्य JSON है, लेकिन यह फ़ाइल को बड़ा कर देता है और बाद में निरीक्षण को कठिन बना देता है।

चरण 4: एक प्रकार की तालिका में रूपांतरित करें

DuckDB JSON लाइनों की फ़ाइल को सीधे पढ़ता है। read_json_auto स्कीमा का अंदाजा लगाता है, और चारों ओर का SELECT वह जगह है जहां आप उन प्रकारों और व्युत्पन्न कॉलमों को लागू करते हैं जो आप वास्तव में चाहते हैं:

import duckdb

con = duckdb.connect("quotes.duckdb")

con.execute("""
    CREATE OR REPLACE TABLE quotes AS
    SELECT
        page::INTEGER            AS page,
        text                     AS quote_text,
        author,
        tags,
        len(tags)                AS tag_count
    FROM read_json_auto('quotes.jsonl')
""")

total = con.sql("SELECT count(*) FROM quotes").fetchone()[0]
authors = con.sql("SELECT count(DISTINCT author) FROM quotes").fetchone()[0]
print(f"लोड की गई पंक्तियाँ: {total} | अलग लेखक: {authors}")

print(con.sql("""
    SELECT author, count(*) AS quotes, round(avg(tag_count), 2) AS avg_tags
    FROM quotes
    GROUP BY author
    ORDER BY quotes DESC, author
    LIMIT 5
""").to_df().to_string(index=False))

con.execute("COPY quotes TO 'quotes.parquet' (FORMAT PARQUET, COMPRESSION ZSTD)")
import os
print(f"पार्केट बाइट्स: {os.path.getsize('quotes.parquet')} | jsonl बाइट्स: {os.path.getsize('quotes.jsonl')}")

rt = duckdb.sql("SELECT count(*) AS n, count(DISTINCT author) AS a FROM 'quotes.parquet'").fetchone()
print(f"पार्केट से राउंड-ट्रिप -> पंक्तियाँ: {rt[0]}, लेखक: {rt[1]}")

लोड की गई पंक्तियाँ: 30 | अलग लेखक: 20
         लेखक  उद्धरण  औसत टैग
Albert Einstein       6      2.83
   J.K. Rowling       3      1.33
     Bob Marley       2      1.00
      Dr. Seuss       2      2.00
 Marilyn Monroe       2      4.00
पार्केट बाइट्स: 4674 | jsonl बाइट्स: 7690
पार्केट से राउंड-ट्रिप -> पंक्तियाँ: 30, लेखक: 20

tags कॉलम एक सूची के रूप में बनी रहती है बजाय इसके कि उसे एक विभाजित स्ट्रिंग में समतल किया जाए। DuckDB नेस्टेड प्रकारों को पार्केट में ले जाती है, इसलिए len(tags) SQL में काम करता है और एरे राउंड ट्रिप से जीवित रहती है — इस चरण में "a,b,c" में समतल करना एक हानिकारक आदत है जिसके टूटने का मूल्य है।

CREATE OR REPLACE TABLE लोड को आइडेम्पोटेंट बनाता है। चरण को फिर से चलाना वर्तमान फ़ाइल से तालिका को पुनर्निर्मित करता है बजाय इसके कि डुप्लिकेट जोड़ने के, जो वह व्यवहार है जो आप चाहते हैं जब आप अभी भी एक्स्ट्रैक्टर पर काम कर रहे हैं।

संक्षेप में यह पूरे अभ्यास का बिंदु है: 30 रिकॉर्ड, 20 अलग लेखक, और एक लेखक जो उनमें से 6 के लिए जिम्मेदार है। यह प्रश्न तालिका के खिलाफ एक एकल पंक्ति क्वेरी है और JSON फ़ाइल के खिलाफ एक निराशाजनक लूप है।

क्या आप इसे उस स्रोत के खिलाफ चलाने के लिए तैयार हैं जो आपको पसंद है? एक मुफ्त Scrapeless खाता बनाएं और फ़ेच स्टेज में URL को बदलें।

चरण 5: पार्केट के रूप में स्टोर करें

एक ही 30 रिकॉर्ड 4,674 बाइट्स के ZSTD-संकुचित पार्केट के रूप में occupy करते हैं जबकि 7,690 बाइट्स JSON लाइनों के रूप में occupy करते हैं। इस नमूने पर बचत मध्यम है; जिस कारण से परवाह करना है वह यह है कि प्रारूप स्केल पर क्या करता है और इसके बाद क्या सक्षम करता है।

पार्केट कॉलम का है और प्रत्येक कॉलम के मानों को अपने स्वयं के एन्कोडिंग के साथ संग्रहीत करता है, जो Apache Parquet फ़ाइल प्रारूप विनिर्देश में परिभाषित है। एक क्वेरी दो में से पांच कॉलमों को छूने पर केवल उन्हीं दो को पढ़ती है। यहां उपयोग किया गया संकुचन कोडेक Zstandard संकुचन मानक में निर्दिष्ट है।

राउंड-ट्रिप लाइन चरण का अपना परीक्षण है। वापस पार्केट फ़ाइल पढ़ने से 30 पंक्तियाँ और 20 अलग लेखक मिलते हैं, जो तालिका के साथ मेल खाते हैं जिसमें से यह आई थी — यह किसी भी पाइपलाइन में_ASSERT करने के लिए मूल्यवान है जो एक फ़ाइल लिखती है जिसे कोई अन्य प्रणाली पढ़ेगी।

ध्यान दें कि अंतिम क्वेरी 'quotes.parquet' को सीधे एक तालिका के रूप में पढ़ती है, बिना किसी आयात चरण और डेटा फ़ाइल के लिए कोई खुला कनेक्शन के। यह पार्केट में स्क्रैपिंग पाइपलाइन समाप्त करने का व्यावहारिक कारण है: आउटपुट डकडबी द्वारा क्वेरी करने योग्य है, और अधिकांश अन्य विश्लेषणात्मक इंजनों द्वारा, जहाँ यह स्थित है उसी स्थान पर।

पूरा पाइपलाइन

एक स्क्रिप्ट के रूप में चलाएं, पांच चरण एक ही पास में पढ़ने के लिए छोटे हैं। यहाँ कॉपी करने के लिए संस्करण है:

import json, os, urllib.request
from html.parser import HTMLParser

API = "https://api.scrapeless.com/api/v2/unlocker/request"

def fetch(url: str) -> str:
    payload = json.dumps({
        "actor": "unlocker.webunlocker",
        "input": {"url": url, "js_render": True, "headless": True},
    }).encode()
    req = urllib.request.Request(
        API, data=payload, 

**Q: क्यूं डकडब की बजाय स्क्रैप की गई डेटा के लिए क्लाउड वेयरहाउस?**

DuckDB एक प्रक्रिया में चलता है, जिसमें कोई सर्वर, कोई खाता और कोई नेटवर्क राउंड ट्रिप नहीं होता, जो उन पैमानों के अनुसार होता है जिन पर अधिकांश स्क्रैपिंग प्रोजेक्ट वास्तव में कार्य करते हैं। यह JSON और Parquet को स्वाभाविक रूप से पढ़ता है, इसलिए कोई लोडर लिखने की आवश्यकता नहीं है। एक क्लाउड गोदाम का स्थान तब होता है जब कई टीमों को एक ही तालिकाओं तक समकालिक पहुंच की आवश्यकता होती है — जब एक पाइपलाइन को अपने स्वयं के आउटपुट पर SQL की आवश्यकता होती है, तब नहीं।

प्रश्न: क्या मुझे लोड करने से पहले टैग सूचियों जैसे निहित फ़ील्ड को समतल करना होगा?

नहीं, और समतल करने से जानकारी खोती है। DuckDB सूची प्रकारों का अंत तक समर्थन करता है, इसलिए tags लोड के दौरान, SQL में len(tags) के माध्यम से, और Parquet लिखने और पढ़ने के दौरान एक ऐरे बना रहता है। इसे एक विभाजित स्ट्रिंग में संकुचित करना हर बाद में आने वाली क्वेरी को फिर से पार्स करने के लिए मजबूर करता है।

प्रश्न: स्क्रैपिंग और लोडिंग के बीच JSON लाइन्स क्यों लिखें?

यह दो हिस्सों को अलग करता है। स्क्रैप वह धीमा, विफलता-प्रवृति भाग है; एक बार रिकॉर्ड डिस्क पर एक-पंक्ति में होने पर, आप लोड और ट्रांसफॉर्म को जितनी बार चाहें फिर से चला सकते हैं बिना दोबारा-fetch किए। पंक्ति दर पंक्ति जोड़ने का अर्थ है कि जो रन आंशिक रूप से मर जाता है, वह पहले के रिकॉर्ड को अक्षुण्ण और पठनीय छोड़ देता है।

प्रश्न: Parquet JSON लाइनों की तुलना में कितनी छोटी है?

इस 30-रेकॉर्ड सैंपल में, 4,674 बाइट की तुलना में 7,690 — लगभग 40% छोटी। इस आकार में उस अनुपात में ज्यादा न पढ़ें, जहां फ़ाइल का ओवरहेड प्रमुख होता है। Parquet का असली लाभ कॉलम रीड्स है: एक क्वेरी जो पांच कॉलम में से दो को छूती है, केवल उन्हीं को पढ़ती है, जो उन मात्रा में महत्वपूर्ण होती है जहां फ़ाइल आराम से मेमोरी में नहीं बैठती।

प्रश्न: क्या मैं Parquet फ़ाइल को पहले एक डेटाबेस में लोड किए बिना क्वेरी कर सकता हूं?

हाँ, और लोड स्टेज की आखिरी पंक्ति ठीक यही करती है — SELECT ... FROM 'quotes.parquet' बिना खुलते डेटाबेस कनेक्शन और बिना आयात के। यही कारण है कि Parquet एक स्क्रैपिंग पाइपलाइन के लिए एक अच्छा अंतिम वस्त्र बना है: आउटपुट वहां बैठकर प्रश्नयोग्य रहता है, DuckDB द्वारा और अन्य विश्लेषणात्मक इंजनों द्वारा।