Implementing Custom Export Pipelines with Scrapling's JSON/JSONL Export: A Complete Guide

Override the on_scraped_item hook in your Scrapling spider to transform, filter, or stream items to custom destinations while retaining the built-in ItemList.to_json() and to_jsonl() methods for final batch exports.

Scrapling provides a high-performance built-in export layer that serializes scraped items to JSON or JSONL formats using the ItemList class. While the default behavior writes the entire collection to disk after the crawl finishes, implementing custom export pipelines with Scrapling's JSON/JSONL export capabilities requires leveraging the on_scraped_item hook to intercept, transform, or redirect individual items before they reach the final collection.

Understanding Scrapling's Built-in Export Architecture

The ItemList Class and JSON/JSONL Methods

The export functionality resides in the ItemList class defined in scrapling/spiders/result.py (lines 10-38). This class maintains an in-memory collection of scraped dictionaries and provides two primary export methods:

• to_json(path, indent=False): Serializes the entire collection to a JSON file using the high-performance orjson library.
• to_jsonl(path): Writes each item as a separate JSON line to a JSONL file, ideal for streaming large datasets.

Both methods handle file I/O and encoding automatically, requiring only a file path argument.

The on_scraped_item Hook for Custom Pipelines

The extensibility point for custom export pipelines is the on_scraped_item hook located in scrapling/spiders/spider.py (lines 86-89). This asynchronous method is invoked immediately after an item is yielded from your parse method but before it is appended to the ItemList.

Method signature:

async def on_scraped_item(self, item: dict) -> dict | None

Return behavior:

• Return the item dict (potentially modified) to include it in the final collection.
• Return None to drop the item entirely, preventing it from reaching the JSON/JSONL export.

Implementing Custom Export Pipelines with on_scraped_item

Transforming Items Before Export

Use the hook to enrich or clean data before it reaches the default JSON export. This is useful for adding timestamps, normalizing fields, or injecting metadata.

from datetime import datetime
from scrapling.spiders import Spider, Response

class TransformSpider(Spider):
    name = "transform_demo"
    start_urls = ["https://example.com/products"]

    async def parse(self, response: Response):
        for product in response.css(".product"):
            yield {
                "name": product.css("h2::text").get(),
                "price": product.css(".price::text").get(),
            }

    async def on_scraped_item(self, item: dict):
        # Enrich with metadata

        item["scraped_at"] = datetime.utcnow().isoformat()
        item["source"] = self.name
        return item

After the crawl completes, result.items.to_json("products.json", indent=True) will include the enriched fields.

Filtering and Dropping Unwanted Items

Implement validation logic to exclude incomplete or invalid records from your JSON/JSONL output by returning None from the hook.

class FilterSpider(Spider):
    name = "filter_demo"
    start_urls = ["https://example.com/listings"]

    async def on_scraped_item(self, item: dict):
        # Drop items missing critical fields

        if not item.get("price") or not item.get("title"):
            return None
        
        # Drop items with invalid price format

        price = item["price"]
        if not isinstance(price, (int, float)) and not price.replace(".", "").isdigit():
            return None
            
        return item

Streaming to Incremental JSONL Files

For large-scale crawls where memory constraints prevent holding all items in ItemList, use on_scraped_item to write incrementally to a JSONL file while the spider runs.

import orjson
from pathlib import Path
from scrapling.spiders import Spider, Response

class StreamSpider(Spider):
    name = "stream_demo"
    start_urls = ["https://quotes.toscrape.com/"]
    concurrent_requests = 5

    async def on_start(self):
        # Initialize file handle once at spider start

        self._jsonl_path = Path("quotes_stream.jsonl")
        self._jsonl_path.parent.mkdir(parents=True, exist_ok=True)
        self._jsonl_file = self._jsonl_path.open("ab")

    async def on_close(self):
        # Ensure file is closed on spider shutdown

        self._jsonl_file.close()

    async def parse(self, response: Response):
        for quote in response.css(".quote"):
            yield {
                "text": quote.css(".text::text").get(),
                "author": quote.css(".author::text").get(),
                "tags": quote.css(".tags .tag::text").getall(),
            }

    async def on_scraped_item(self, item: dict):
        # Serialize and write immediately using orjson for speed

        self._jsonl_file.write(orjson.dumps(item, option=orjson.OPT_SERIALIZE_NUMPY))
        self._jsonl_file.write(b"\n")
        return item  # Optional: keep in memory for final dump if needed

This approach leverages orjson (the same library used by ItemList) for consistent performance while bypassing the in-memory collection.

Integrating with External Message Queues

Extend the pattern to send items to RabbitMQ, Redis, or other message brokers instead of (or in addition to) local JSON files.

import aio_pika
import orjson
from scrapling.spiders import Spider

class QueueSpider(Spider):
    name = "queue_demo"
    start_urls = ["https://example.com/feed"]

    async def on_start(self):
        self._connection = await aio_pika.connect_robust("amqp://guest:guest@localhost/")
        self._channel = await self._connection.channel()
        self._queue = await self._channel.declare_queue("scrapling_items")

    async def on_close(self):
        await self._connection.close()

    async def on_scraped_item(self, item: dict):
        # Publish to RabbitMQ

        await self._channel.default_exchange.publish(
            aio_pika.Message(body=orjson.dumps(item)),
            routing_key=self._queue.name
        )
        # Return None to prevent storing in ItemList (memory optimization)

        return None

Combining Default Exports with Custom Pipelines

You can simultaneously use on_scraped_item for real-time processing and retain the built-in ItemList export for a final snapshot. This hybrid approach is useful when you need both a live stream and a complete batch file.

class HybridSpider(Spider):
    name = "hybrid"
    start_urls = ["https://example.com/data"]

    async def on_scraped_item(self, item: dict):
        # Real-time validation

        if not item.get("id"):
            return None
        
        # Enrich data

        item["processed"] = True
        return item  # Keeps item in ItemList

# After crawl:

# 1. Real-time processing handled in hook

# 2. Final export available via result.items.to_json("final.json")

Summary

• ItemList in scrapling/spiders/result.py provides high-performance batch export via to_json() and to_jsonl() using the orjson library.
• on_scraped_item in scrapling/spiders/spider.py is the asynchronous hook called after each item is yielded but before it reaches the ItemList, enabling real-time transformation, filtering, or streaming.
• Return None from on_scraped_item to drop items entirely, preventing them from appearing in JSON/JSONL exports.
• Combine approaches by using on_scraped_item for custom pipelines (databases, message queues, incremental files) while still calling result.items.to_json() for a final batch snapshot.
• Memory optimization for large crawls is achieved by streaming items to disk in on_scraped_item and returning None to avoid populating the in-memory ItemList.

Frequently Asked Questions

How do I drop items that fail validation before they reach the JSON export?

Override the on_scraped_item method in your spider class and return None for invalid items. According to the implementation in scrapling/spiders/spider.py (lines 86-89), returning None prevents the item from being appended to the ItemList, effectively excluding it from both the in-memory collection and any subsequent calls to to_json() or to_jsonl().

Can I use both custom streaming pipelines and the default batch export simultaneously?

Yes. When you return the item dict from on_scraped_item (instead of None), the item continues to the ItemList as usual. You can therefore write custom logic inside the hook to stream items to a database or message queue in real-time, then call result.items.to_json("final.json") after start() completes to obtain a complete batch file of all retained items.

What is the performance impact of using on_scraped_item for file I/O?

The hook is asynchronous, so you can use await with non-blocking I/O operations. For maximum performance when writing to JSONL files, use the orjson library (the same serializer used by ItemList) and open the file in binary append mode ("ab") as shown in the streaming example. This approach minimizes memory overhead and maintains high throughput even for millions of items.

How do I handle resource cleanup when using custom pipelines?

Implement the on_start and on_close lifecycle hooks available in the base Spider class. Initialize file handles, database connections, or message queue clients in on_start, and ensure proper cleanup in on_close. This guarantees that resources are released correctly even if the spider is interrupted or encounters an error during the crawl.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}