Skip to content

Appendix G: Issuer Classes

Jump to bottom
Martin Olveyra edited this page Jul 15, 2026 · 9 revisions

Previous Chapter: Appendix F: Monitor Classes

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 (or, when explode_input_items is set, once per object the jmespath selects inside each 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). The matched spiders are visited round-robin, rotating the starting spider across loops (only max_inputs_per_loop inputs run per loop, and each loop resumes where the previous one stopped), so when the target matches many spiders — e.g. class:<BaseSpider> to consume every source in one job — a source with a large backlog cannot starve the others; it is a no-op for a single-spider target. It also exposes a couple of extra flags — see the next section.

IssuerScriptWithSCJobInput options

Beyond the generic issuer attributes, IssuerScriptWithSCJobInput adds two flags:

Attribute Default Meaning
set_item_source True Stamp each read item's source with the scanned spider's canonical name. Set to False when the scanned spider is a secondary / post-processing stage — one that processes data already produced by a primary spider. Such items already carry their originating source, which must be conserved rather than overwritten. (This is exactly why a delivery issuer, which reads a secondary spider, keeps the upstream source.)
flush_on_each_input False When True, flush_files() runs at the end of processing each scanned job (instead of only when a source's spider stops or on close). Some pipelines instead deliver several jobs into one output batch and leave this False.
scope_input_to_flow_id False When True, read only input jobs tagged with this script's own FLOW_ID (from --flow-id, or the FLOW_ID tag when running inside a workflow), so a script scheduled by a graph manager reads all and only the jobs of its own workflow instance. Ports the workflow-scoping the deprecated BaseDeliverScript did automatically. No-op (reads everything, with a warning) if no flow_id is set.

One output file per scanned job. flush_on_each_input guarantees a flush per job, but a batch can still be split earlier if it reaches default_filesize. To ensure all items of a single job land in the same output file, pair flush_on_each_input=True with a large default_filesize (or a get_filesize_from_item() returning a big number for the relevant items).

process_input() also calls a post_process_input_items(spider_job, args) hook (default a no-op) once per scanned job, after all of that job's items have been read (and processed via process_item()) and before the optional per-job flush. spider_job is the just-read Scrapy Cloud job — use its metadata / key / items (the key is also args[0]["key"]). Override it to run any per-job finalization logic.

Its most common use is the accumulate-then-merge pattern, which is generic to any IssuerScriptWithSCJobInput (delivery or not): when an input's records must be combined into fewer output records (a join, roll-up, or reconciliation), override process_item() to accumulate the records instead of issuing them inline; then in post_process_input_items() merge them, set each combined record's id / input_source, issue_item() the results, and reset the accumulator. Paired with flush_on_each_input=True, the merged records are written as a single output file per job. A delivery issuer may use this pattern, but so may a filter or a roll-up stage — it is not a property of delivery.

The hook is also the place to aggregate a scanned job's stats if you need it: there is no built-in flag for that — mix in SpiderStatsAggregatorMixin yourself and call self.aggregate_spider_stats(...) from your post_process_input_items() override.

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).
separate_output_by_source True one output file per (slot, source). False ⇒ all sources of a slot share one queue and pack into default_filesize-sized files (one file per slot), the source is dropped from the default filename, and the per-source stopped-spider partial flush is disabled (a slot's batch flushes only at default_filesize or on close).
persist_items_queue_on_disk False each output-queue bucket (items_queue[slot][source]) is an in-memory dict by default — fast, but a whole batch is held in RAM until flushed. True ⇒ back each bucket with an on-disk SqliteDict: items live on disk and are streamed to the output file at flush, so memory stays bounded (slower). Use it when items are large enough (e.g. a delivery of records carrying heavy metadata) that a batch would not fit in memory.
explode_input_items None a jmespath expression selecting a list inside each raw input record; when set, process_item() is called once per selected object (one record → many items) instead of once per record. Lets a spider whose records bundle several issuable objects be consumed without overriding process_input().
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 issuer. 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.

There is no dedicated delivery class. A delivery is simply an IssuerScriptWithSCJobInput configured for the terminal stage — the base class already provides everything a delivery needs. Which attributes you set is use-case dependent, not fixed by "delivery":

  • close_on_no_inputs = True — typical: deliver the finished jobs, then stop rather than loop forever. Override for a continuously-running delivery.
  • flush_on_each_input = True + a large default_filesize — when you want one delivery file per scanned job (the common case). Leave flush_on_each_input at False if a delivery batches several jobs into one output file.
  • dedupe = False — set this only when the delivered data was already deduplicated upstream (a post-processing pipeline: consumer → deduplicator → … → delivery). A delivery that reads a primary spider's crawl directly may keep dedupe = True (or reproduce the old field-based dedup).
  • set_item_source = False — set this only when the delivered spider is a secondary / post-processing stage whose items already carry their originating source. Delivering a primary spider keeps the default (True), so source is stamped with that spider's canonical name.

You always supply:

  • build_item_id() — required (abstract); issue_item() uses the id for output queueing/routing even when dedupe = False.
  • the delivery destination — via output_folder and/or a compute_destination_filename(output_slot, source) override. With flush_on_each_input = True there is a natural per-job → per-file mapping; the destination is often derived from the scanned job's spider args (read them in post_process_input_items()).

Delivery issues each read item as-is by default. Combining records before writing is not a property of delivery: if a particular delivery needs it, it uses the generic accumulate-then-merge pattern via post_process_input_items(), exactly as any other issuer would.

Migrating from BaseDeliverScript

Rebuild the delivery as an IssuerScriptWithSCJobInput (no dedicated class) and choose the configuration above to fit your pipeline. The old hooks map over as follows:

  • on_item(item, scrapername) → process_item(item, input_source) (issue inline, or accumulate);
  • process_job_items(scrapername, job) → the base process_input() reads the job, and per-job finalization goes in post_process_input_items(spider_job, args);
  • uploading a per-job file via the filesystem helper → issue_item() + a per-job flush (flush_on_each_input = True), with the path from compute_destination_filename();
  • DEDUPE_KEY_BY_FIELDS / is_seen_item() → dedupe + build_item_id() (bloom filter);
  • the scrapername positional arg → the target arg (spider: / canonical: / class:);
  • DELIVERED_TAG = "delivered" (which marked done jobs) → the issuer selects jobs lacks_tag=CONSUMED_TAG and tags consumed ones with CONSUMED_TAG (default "CONSUMED=True").

Watch out for: preserving the "delivered" tag (production-critical) — BaseDeliverScript tagged delivered jobs "delivered", but the issuer defaults to CONSUMED_TAG = "CONSUMED=True"; deploying the migration with the default would re-deliver every already-delivered job (none carry CONSUMED=True), so set CONSUMED_TAG = "delivered" on the migrated issuer; the invocation changes (deliver.py <scrapername…> → deliver.py <type>:<name>, so update the scheduling job); FLOW_ID scoping — the issuer reads all finished, un-consumed jobs of the target, so if a graph manager used to scope delivery to its workflow instance, set scope_input_to_flow_id = True (which the deprecated class did automatically); conserving the source for a secondary spider (set_item_source = False, or the base overwrites item["source"]); reading the per-job destination from the job's spider args in post_process_input_items(); and converting scrapy.Items to plain dicts before issuing. For per-job stats aggregation there is no built-in flag — mix in SpiderStatsAggregatorMixin and call aggregate_spider_stats(...) in the hook.

💡 Claude assistance. shub-workflow ships a Claude Code plugin, shub-workflow-toolkit, whose shub-workflow-issuers skill teaches Claude how to build issuers and carry out this BaseDeliverScript → issuer migration (it bundles a dedicated migration reference covering the configuration choices, method mapping and gotchas above). Install it as described in the repository README and Claude will be able to assist you with the migration.

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.

Tutorial TOC

Using Frontera (old approach, less recommended)

Appendices

Clone this wiki locally