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

> Master custom export pipelines in Scrapling using the on_scraped_item hook. Transform, filter, and stream data efficiently with JSON/JSONL export.

- Repository: [Karim shoair/Scrapling](https://github.com/D4Vinci/Scrapling)
- Tags: how-to-guide
- Published: 2026-03-08

---

**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`](https://github.com/D4Vinci/Scrapling/blob/main/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`](https://github.com/D4Vinci/Scrapling/blob/main/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:**

```python
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.

```python
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.

```python
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.

```python
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.

```python
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.

```python
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`](https://github.com/D4Vinci/Scrapling/blob/main/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`](https://github.com/D4Vinci/Scrapling/blob/main/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`](https://github.com/D4Vinci/Scrapling/blob/main/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.