Appendix G: Issuer Classes

Previous Chapter: Appendix F: Monitor Classes

---

• Introduction
• Class hierarchy
• The item model
• IssuerScript
• Output: slots, batches, filenames
• Deduplication
• The two ready-made input subclasses
• Configuration attributes
• Delivery as an issuer
• Writing and running an issuer

Introduction

An issuer is a fundamental Scrapy Cloud workflow component: it reads a massive input (finished spider jobs, or batch files in a folder), processes the items (dedup, filter, transform, split, re-tag…), and writes a massive output as batch files. Chaining issuers builds the whole post-crawl data-processing pipeline — consumers, deduplicators, balancers, and final delivery are all issuers. IssuerScript (shub_workflow/issuer/__init__.py) is the base class.

It is a BaseLoopScript (see Appendix B: Script Classes): it loops, reading new inputs and flushing output batches, until its inputs are exhausted (or forever, if it's a continuous stage). It is generic over two type parameters: the item type (ITEMTYPE, a subtype of IssuerItem) and the per-input args passed from input discovery to input processing (PROCESS_INPUT_ARGS_TYPE).

Issuers also supersede the old delivery base class — see Delivery as an issuer.

Class hierarchy

BaseLoopScript                          (Appendix B)
      ▲
IssuerScript[ITEMTYPE, PROCESS_INPUT_ARGS_TYPE]     read input → process/dedup → write batch output
      ▲
      ├── IssuerScriptWithFileSystemInput[ITEMTYPE]        input = batch files in a folder
      └── IssuerScriptWithSCJobInput[ITEMTYPE]             input = finished Scrapy Cloud spider jobs

The item model

Items flowing through an issuer are IssuerItem TypedDicts:

Field	Meaning
id	unique id, computed by your build_item_id() and used for dedup.
source	the source crawler / canonical name the item belongs to.
input_source	where the item was read from (a job key or input file name).
search_keywords	(optional) set of keywords the item came from; merged when the same id recurs.

Subclass IssuerItem to add your own fields and bind the issuer to it (class MyIssuer(IssuerScript[MyItem, Tuple[()]])). You must implement build_item_id(item) -> ItemId. Override adapt_input_item(raw) -> ITEMTYPE if the raw input records aren't already in your item shape (default just casts).

IssuerScript

The loop & data flow

workflow_loop() (provided) calls get_new_inputs() → for each input, process_input(), which iterates the input's records and calls process_item() per record. process_item():

1. stamps input_source, computes id = build_item_id(item), counts totals;
2. if dedup is on and the id was already seen, drops it (counts a dupe);
3. otherwise issue_item() — enqueue into the in-memory items_queue[slot][source], and when that queue reaches the item's filesize, send_file() flushes it to a gzipped JSON-lines batch file.

On close it flushes remaining queues and removes/marks the fully-processed inputs. An input is only removed once all the items it contributed have been written out (tracked in pending_inputs_to_remove).

What you implement

Method	Required?	Purpose
build_item_id(item) -> ItemId	abstract	the dedup/identity key for an item.
get_new_inputs() -> Iterable[(InputSource, args)]	abstract*	discover inputs to process (the ready subclasses implement this).
process_input(inputsrc, args) -> bool	abstract*	read one input, call process_item() per record; return True if processed.
remove_inputs(inputs)	abstract*	consume/retire processed inputs (the subclasses implement this).
adapt_input_item(raw)	optional	adapt raw records to your ITEMTYPE.
get_output_slot_for_item / get_filesize_from_item / compute_destination_filename	optional	customize output routing / batch size / naming.

* implemented for you by the two input subclasses below; you usually only write build_item_id (and a custom IssuerItem) plus set a few attributes.

Output: slots, batches, filenames

Items are queued per (output slot, source) and flushed to a file when the queue reaches the item's filesize (default_filesize, or per-item via get_filesize_from_item). Output goes to output_folder; filenames are timestamped and prefixed by source (and slot / input_slot when set), e.g. <output_slot>_<source>_<YYYYmmddThhmmss.ffffff>_<input_slot>.jl.gz.

Output slots enable parallel downstream processing: set parallel_outputs = N and each item is routed to slot hash_mod(id, N) (so the same id always lands in the same slot — letting N parallel downstream issuers each own a slot). Or pin a single output_slot. input_slot similarly selects one input slot for this instance (N parallel instances, one per slot).

Deduplication

When dedupe is on (default), process_item() skips ids already in a bloom filter (seen, capacity MAX_ITEMS, error rate ERRORS_RATE, persisted to a livedup.bloom file). Plain in-memory dedup only covers the job's own lifetime — fine for a short-lived stage, lost on restart.

For long-term dedup that survives restarts, set LOAD_DELIVERED_IDS_DAYS and call load_last_outputs(output_folders, …) in __init__ — on start it reads the output files (selected by their timestamped names) from the last N days back into seen. output_folders is a tuple, so it can include downstream folders (where items land after further processing), not just this issuer's own output — ensuring an id already passed along stays deduplicated. This is what makes a deduplicator's dedup persistent across restarts (see Consumers and deduplicators). If you set LOAD_DELIVERED_IDS_DAYS but never call load_last_outputs, on_start() raises (a guard against silently-incomplete dedup). Bloom filters have a small false-positive rate (a unique item dropped as a dupe) — acceptable for throughput-oriented pipelines.

The two ready-made input subclasses

Pick one; it implements get_new_inputs / process_input / remove_inputs so you only write build_item_id (+ output attrs):

• IssuerScriptWithFileSystemInput — input is batch files in input_folder (optionally filtered by input_slot prefix). Each file is a gzipped JSON-lines list; processed inputs are moved to processed_folder if set, else removed. The typical mid-pipeline stage (consumer → filter → deduplicator → balancer).
• IssuerScriptWithSCJobInput — input is finished Scrapy Cloud spider jobs. A required positional CLI arg target (spider:<name> | canonical:<name> | class:<ClassName>) selects which spiders' jobs to read; their items are processed and the consumed jobs are tagged CONSUMED=True (so they aren't re-read). The typical first stage (reading raw crawl output).

Consumers and deduplicators

Two issuer archetypes recur, making opposite trade-offs around dedup persistence:

• Consumer — the first stage. Reads spider jobs, does a cheap, in-memory preliminary dedup, and distributes items across parallel_outputs slots (the same id always lands in the same slot, so dedup can be parallelized downstream), often also extracting seeds to the frontier. A consumer is short-lived and cheaply restartable: it does not use load_last_outputs, so a restart loses only its in-memory dedup — acceptable, because the authoritative dedup is downstream. It's the right place for post-processing logic that changes over time.
• Deduplicator — the heavy stage. Runs continuously, typically one instance per slot, doing the massive, long-term deduplication. It survives restarts: on start it calls load_last_outputs(...) to refill its bloom filter from the last LOAD_DELIVERED_IDS_DAYS of its own (and downstream) output, so previously-emitted ids stay deduplicated. Deduplicators usually have a per-instance capacity (MAX_ITEMS) and run until it's reached.

Why not do the massive dedup in the consumer? Two reasons. (1) The consumer is a single process that also writes seeds and holds the post-processing logic that changes over time, so it must be restartable cheaply — doing the persistent massive dedup there would make every restart reload an enormous id history (and risk the dedup). (2) Massive dedup must scale beyond one process: the consumer instead fans items out by id-hash across parallel_outputs slots, so N deduplicators each dedup one slot independently (scale further by adding slots). Keeping them separate lets the volatile consumer be stopped/updated/rescheduled without affecting the authoritative, persistent deduplication.

Configuration attributes

Attribute	Default	Meaning
output_folder	(required)	where batch output files are written.
default_filesize	10_000	items per output batch file.
parallel_outputs	1	number of output slots (>1 ⇒ hash-routed slots).
input_slot / output_slot	None	pin this instance to one input / output slot.
dedupe	True	enable bloom-filter de-duplication.
MAX_ITEMS	200_000_000	bloom capacity; also the loop's processed-items ceiling.
ERRORS_RATE	1e-8	bloom false-positive rate.
LOAD_DELIVERED_IDS_DAYS	(unset)	days of prior output to reload into seen (needs load_last_outputs).
close_on_no_inputs	False	stop when there are no more inputs (vs. keep looping).
max_inputs_per_loop	-1	cap inputs processed per loop cycle (-1 = unlimited).
min_wait_time_secs_to_flush_stopped_spiders	0	grace before flushing pending items of a source whose spider has stopped.
loop_mode	0	(inherited) seconds between cycles; set it for a continuous issuer.

Coupling note: each loop, if a source still has queued items, the issuer checks get_project_running_spiders(crawlmanagers=("py:crawlmanager.py",)); if that source's spider is no longer running it flushes the partial batch after min_wait_time_secs_to_flush_stopped_spiders. So the default expects the upstream discovery crawl manager to be named py:crawlmanager.py.

Delivery as an issuer

The final delivery stage is now built as an IssuerScript (a last-stage issuer that writes the customer's delivery files). This replaces the older shub_workflow.deliver.BaseDeliverScript, which is deprecated — instantiating it now emits a DeprecationWarning. New delivery scripts, and migrations of old ones, should be issuers.

Writing and running an issuer

A minimal file-input issuer (a mid-pipeline consumer/deduplicator):

import logging
from typing import Tuple
from shub_workflow.issuer import IssuerScript, IssuerScriptWithFileSystemInput, IssuerItem, ItemId


class MyItem(IssuerItem):
    url: str


class MyIssuer(IssuerScriptWithFileSystemInput[MyItem]):

    loop_mode = 60
    input_folder = "gs://my-bucket/stage-in"
    output_folder = "gs://my-bucket/stage-out"
    parallel_outputs = 10                 # 10 hash-routed output slots
    LOAD_DELIVERED_IDS_DAYS = 30          # dedup against the last 30 days of output

    def __init__(self):
        super().__init__()
        self.load_last_outputs((self.output_folder,))   # required when LOAD_DELIVERED_IDS_DAYS is set

    def build_item_id(self, item: MyItem) -> ItemId:
        return ItemId(item["url"])


if __name__ == "__main__":
    from shub_workflow.utils import get_kumo_loglevel

    logging.basicConfig(format="%(asctime)s %(name)s [%(levelname)s]: %(message)s", level=get_kumo_loglevel())
    MyIssuer().run()

As with any shub-workflow script, register the file in your project's setup.py so it deploys and runs on Scrapy Cloud as py:myissuer.py; deployment is out of scope here.