diff --git a/VERSION b/VERSION
index f90b1af..0bee604 100644
--- a/VERSION
+++ b/VERSION
@@ -1 +1 @@
-2.3.2
+2.3.3
diff --git a/metrics/counter/access/accumulation.py b/metrics/counter/access/accumulation.py
index ae0b6c8..3f6a616 100644
--- a/metrics/counter/access/accumulation.py
+++ b/metrics/counter/access/accumulation.py
@@ -28,19 +28,10 @@ def accumulate(results, counter_access, line):
         access_datetime.date().toordinal(),
         access_datetime.hour,
     )
-    compact_accumulate = getattr(results, "accumulate_access", None)
-    user_session_id = None
-    if compact_accumulate is None:
-        user_session_id = _generate_user_session_id(
-            client_name, client_version, ip_address, access_datetime
-        )
     raw_record = _build_record(
         counter_access=counter_access,
         line=line,
         access_datetime=access_datetime,
-        second_of_hour=second_of_hour,
-        user_session_id=user_session_id,
-        include_id=compact_accumulate is None,
     )
     access_url_key = access_url or "|".join(
         [
@@ -50,33 +41,18 @@ def accumulate(results, counter_access, line):
         ]
     )
 
-    if compact_accumulate is not None:
-        compact_accumulate(
-            data=raw_record["data"],
-            session_key=session_key,
-            url=access_url_key,
-            second=second_of_hour,
-        )
-        return
-
-    item_access_id = raw_record["id"]
-    if item_access_id not in results:
-        results[item_access_id] = raw_record["data"]
-
-    timestamps_by_url = results[item_access_id].setdefault(
-        "click_timestamps_by_url", {}
+    results.accumulate_access(
+        data=raw_record,
+        session_key=session_key,
+        url=access_url_key,
+        second=second_of_hour,
     )
-    url_timestamps = timestamps_by_url.setdefault(access_url_key, {})
-    _increment_timestamp_count(url_timestamps, second_of_hour)
 
 
 def _build_record(
     counter_access,
     line,
     access_datetime,
-    second_of_hour,
-    user_session_id,
-    include_id=True,
 ):
     collection = counter_access.get("collection")
     source_key = _source_key(counter_access, collection)
@@ -89,51 +65,27 @@ def _build_record(
     access_country_code = line.get("country_code")
     access_date = access_datetime.strftime("%Y-%m-%d")
 
-    record = {
-        "data": {
-            "collection": collection,
-            "source_key": source_key,
-            "document_type": counter_access.get("document_type"),
-            "pid_v2": pid_v2,
-            "pid_v3": pid_v3,
-            "pid_generic": pid_generic,
-            "document": _document_metadata(counter_access),
-            "title_pid_generic": counter_access.get("title_pid_generic") or pid_generic,
-            "user_session_id": user_session_id,
-            "click_timestamps_by_url": {},
-            "media_format": media_format,
-            "content_language": content_language,
-            "content_type": content_type,
-            "access_country_code": access_country_code,
-            "access_date": access_date,
-            "access_year": access_date[:4],
-            "access_month": access_date[:7].replace("-", ""),
-            "publication_year": counter_access.get("publication_year"),
-            "counter_access_type": counter_access.get("counter_access_type") or "Open",
-            "access_method": counter_access.get("access_method") or "Regular",
-            "source": _source_metadata(counter_access),
-        },
+    return {
+        "collection": collection,
+        "source_key": source_key,
+        "document_type": counter_access.get("document_type"),
+        "pid_v2": pid_v2,
+        "pid_v3": pid_v3,
+        "pid_generic": pid_generic,
+        "document": _document_metadata(counter_access),
+        "title_pid_generic": counter_access.get("title_pid_generic") or pid_generic,
+        "media_format": media_format,
+        "content_language": content_language,
+        "content_type": content_type,
+        "access_country_code": access_country_code,
+        "access_date": access_date,
+        "access_year": access_date[:4],
+        "access_month": access_date[:7].replace("-", ""),
+        "publication_year": counter_access.get("publication_year"),
+        "counter_access_type": counter_access.get("counter_access_type") or "Open",
+        "access_method": counter_access.get("access_method") or "Regular",
+        "source": _source_metadata(counter_access),
     }
-    if include_id:
-        record["id"] = _generate_item_access_id(
-            user_session_id=user_session_id,
-            col_acron3=collection,
-            source_key=source_key,
-            pid_v2=pid_v2,
-            pid_v3=pid_v3,
-            pid_generic=pid_generic,
-            content_language=content_language,
-            access_country_code=access_country_code,
-            media_format=media_format,
-            content_type=content_type,
-        )
-    return record
-
-
-def _increment_timestamp_count(timestamps, key):
-    if key not in timestamps:
-        timestamps[key] = 0
-    timestamps[key] += 1
 
 
 def _normalized_access_path(url):
@@ -150,23 +102,6 @@ def _normalized_access_path(url):
     return path or None
 
 
-def _generate_user_session_id(
-    client_name, client_version, ip_address, datetime, sep="|"
-):
-    dt_year_month_day = datetime.strftime("%Y-%m-%d")
-    dt_hour = datetime.strftime("%H")
-
-    return sep.join(
-        [
-            str(client_name),
-            str(client_version),
-            str(ip_address),
-            str(dt_year_month_day),
-            str(dt_hour),
-        ]
-    )
-
-
 def _document_metadata(counter_access):
     document_title = counter_access.get("document_title")
     return {"title": document_title} if document_title else {}
@@ -196,32 +131,3 @@ def _source_key(counter_access, fallback):
         or counter_access.get("source_type")
         or fallback
     )
-
-
-def _generate_item_access_id(
-    col_acron3,
-    source_key,
-    pid_v2,
-    pid_v3,
-    pid_generic,
-    user_session_id,
-    access_country_code,
-    content_language,
-    media_format,
-    content_type,
-    sep="|",
-):
-    return sep.join(
-        [
-            col_acron3,
-            str(source_key or ""),
-            pid_v2 or "",
-            pid_v3 or "",
-            pid_generic or "",
-            str(user_session_id or ""),
-            str(access_country_code or ""),
-            str(content_language or ""),
-            str(media_format or ""),
-            str(content_type or ""),
-        ]
-    )
diff --git a/metrics/counter/access/daily_accumulator.py b/metrics/counter/access/daily_accumulator.py
index 1cceac7..d376765 100644
--- a/metrics/counter/access/daily_accumulator.py
+++ b/metrics/counter/access/daily_accumulator.py
@@ -117,11 +117,11 @@ def _timestamps_as_dict(self, accumulator):
         return timestamps
 
 
-class DailyAccessAccumulator(dict):
+class DailyAccessAccumulator:
     """Store compact records and materialize them only for metric conversion."""
 
     def __init__(self):
-        super().__init__()
+        self._records = {}
         self._documents = [None, {}]
         self._document_ids = {}
         self._sources = [None]
@@ -130,28 +130,8 @@ def __init__(self):
         self._strings = [None]
         self._string_ids = {}
 
-    def __setitem__(self, key, value):
-        if isinstance(value, _CompactAccessRecord):
-            super().__setitem__(key, value)
-            return
-
-        source_key = value.get("source_key")
-        source = value.get("source")
-        if source_key and source:
-            source_id = self._intern_source(source_key, source)
-            value["source"] = self._sources[source_id]
-
-        document_key = self._document_key(value)
-        document = value.get("document")
-        if document is not None and any(document_key[1:]):
-            document_id = self._intern_document(value)
-            value["document"] = self._documents[document_id]
-
-        user_session_id = value.get("user_session_id")
-        if user_session_id:
-            value["user_session_id"] = self._legacy_intern_session(user_session_id)
-
-        super().__setitem__(key, value)
+    def __len__(self):
+        return len(self._records)
 
     def accumulate_access(self, data, session_key, url, second):
         session = self._intern_session(session_key)
@@ -167,19 +147,35 @@ def accumulate_access(self, data, session_key, url, second):
             self._intern(data.get("media_format")),
             self._intern(data.get("content_type")),
         )
-        record = dict.get(self, key)
+        record = self._records.get(key)
         if record is None:
             record = _CompactAccessRecord(self, data, session, url, second)
-            dict.__setitem__(self, key, record)
+            self._records[key] = record
             return
         record.add_timestamp(self._intern(url), second)
 
-    def iter_materialized_values(self):
-        for value in dict.values(self):
-            if isinstance(value, _CompactAccessRecord):
+    def iter_materialized_values(self, consume=False):
+        if not consume:
+            for value in self._records.values():
                 yield value.as_dict(self)
-            else:
-                yield value
+            return
+
+        keys = tuple(self._records)
+        try:
+            for key in keys:
+                yield self._records.pop(key).as_dict(self)
+        finally:
+            self.clear()
+
+    def clear(self):
+        self._records.clear()
+        self._documents.clear()
+        self._document_ids.clear()
+        self._sources.clear()
+        self._source_ids.clear()
+        self._sessions.clear()
+        self._strings.clear()
+        self._string_ids.clear()
 
     def _intern(self, value):
         if value is None:
@@ -208,13 +204,6 @@ def _intern_session(self, session_key):
             self._sessions[compact_key] = session_id
         return session_id
 
-    def _legacy_intern_session(self, session):
-        interned = self._sessions.get(session)
-        if interned is None:
-            self._sessions[session] = session
-            return session
-        return interned
-
     def _intern_source(self, source_key, source):
         if source is None:
             return _NONE_METADATA_ID
diff --git a/metrics/counter/indexing/converter.py b/metrics/counter/indexing/converter.py
index 5ad0d39..d6a1653 100644
--- a/metrics/counter/indexing/converter.py
+++ b/metrics/counter/indexing/converter.py
@@ -14,22 +14,11 @@
 _DEFAULT = DocumentPipeline()
 
 
-def convert(data):
-    if not isinstance(data, dict):
-        return {"month": {}, "year": {}}
-
-    month_data = _convert_granularity(data, "month")
-    year_data = _convert_granularity(data, "year")
-
-    return {"month": month_data, "year": year_data}
-
-
-def _convert_granularity(data, granularity):
+def convert_granularity(values, granularity):
     converted_data = {}
     unique_state = _initialize_unique_state()
-    values = getattr(data, "iter_materialized_values", data.values)
 
-    for value in values():
+    for value in values:
         pipeline = _get_pipeline(value)
         pipeline.accumulate(
             data=converted_data,
diff --git a/metrics/opensearch/client.py b/metrics/opensearch/client.py
index 271acee..9dd2fb3 100644
--- a/metrics/opensearch/client.py
+++ b/metrics/opensearch/client.py
@@ -10,10 +10,13 @@
     merge_metric_document,
 )
 
+_BULK_CHUNK_SIZE = 500
+
 
 class OpenSearchUsageClient:
     def __init__(self, url=None, basic_auth=None, api_key=None, verify_certs=None):
         self.client = self.get_opensearch_client(url, basic_auth, api_key, verify_certs)
+        logging.info("OpenSearch HTTP request compression is enabled.")
 
     def get_opensearch_client(
         self,
@@ -33,10 +36,20 @@ def get_opensearch_client(
                 url,
                 http_auth=tuple(basic_auth),
                 verify_certs=verify_certs,
+                http_compress=True,
             )
         if api_key:
-            return OpenSearch(url, api_key=api_key, verify_certs=verify_certs)
-        return OpenSearch(url, verify_certs=verify_certs)
+            return OpenSearch(
+                url,
+                api_key=api_key,
+                verify_certs=verify_certs,
+                http_compress=True,
+            )
+        return OpenSearch(
+            url,
+            verify_certs=verify_certs,
+            http_compress=True,
+        )
 
     def ping(self):
         try:
@@ -104,20 +117,17 @@ def index_documents(self, index_name, documents, ping_client=False):
             ),
         )
 
-    def increment_documents_for_daily_job(
+    def increment_document_items_for_daily_job(
         self,
         index_name,
-        documents,
+        document_items,
         job_id,
         ping_client=False,
     ):
         if ping_client and not self.ping():
             return
 
-        if not documents:
-            return
-
-        helpers.bulk(
+        succeeded, _failed = helpers.bulk(
             self.client,
             (
                 build_idempotent_job_increment_action(
@@ -126,9 +136,11 @@ def increment_documents_for_daily_job(
                     document=document,
                     job_id=job_id,
                 )
-                for doc_id, document in documents.items()
+                for doc_id, document in document_items
             ),
+            chunk_size=_BULK_CHUNK_SIZE,
         )
+        return succeeded
 
     def delete_documents(self, index_name, doc_ids, ping_client=False):
         if ping_client and not self.ping():
diff --git a/metrics/services/daily_metric_exports.py b/metrics/services/daily_metric_exports.py
index 7a14a21..52965f3 100644
--- a/metrics/services/daily_metric_exports.py
+++ b/metrics/services/daily_metric_exports.py
@@ -1,11 +1,12 @@
 import logging
+import resource
 from time import monotonic
 
 from metrics.models import DailyMetricJob
 from metrics.opensearch.client import OpenSearchUsageClient
 from metrics.services.export import (
+    daily_metric_payload_exists,
     export_daily_metric_payload,
-    load_daily_metric_payload,
 )
 from metrics.services.jobs import (
     acquire_daily_metric_job,
@@ -27,12 +28,12 @@ def build_and_export_daily_metric_job(job_id, track_errors=False, robots_source=
         return
 
     try:
-        payload = _load_or_build_payload(
+        _ensure_payload(
             job=job,
             track_errors=track_errors,
             robots_source=robots_source,
         )
-        _export_payload(job=job, payload=payload)
+        _export_payload(job=job)
     except Exception as exc:
         logging.error("Failed to process daily metric job %s: %s", job_id, exc)
         mark_daily_metric_job_failed(job, exc)
@@ -41,26 +42,29 @@ def build_and_export_daily_metric_job(job_id, track_errors=False, robots_source=
     mark_daily_metric_job_exported(job)
 
 
-def _load_or_build_payload(job, track_errors, robots_source):
-    payload = load_daily_metric_payload(job)
-    if payload is not None and job.payload_hash:
-        return payload
+def _ensure_payload(job, track_errors, robots_source):
+    if job.payload_hash and daily_metric_payload_exists(job):
+        logging.info(
+            "Daily metric job %s is resuming from persisted payload %s.",
+            job.pk,
+            job.storage_path,
+        )
+        return
 
     robots_list, mmdb = fetch_required_resources(robot_source=robots_source)
     if not robots_list or not mmdb:
         raise RuntimeError("Required parsing resources are not available.")
 
-    payload = build_daily_metric_job_payload(
+    build_daily_metric_job_payload(
         job=job,
         robots_list=robots_list,
         mmdb=mmdb,
         track_errors=track_errors,
     )
     job.refresh_from_db()
-    return payload
 
 
-def _export_payload(job, payload):
+def _export_payload(job):
     opensearch_started = monotonic()
     search_client = OpenSearchUsageClient()
     if not search_client.ping():
@@ -69,10 +73,15 @@ def _export_payload(job, payload):
     export_daily_metric_payload(
         search_client=search_client,
         job=job,
-        payload=payload,
     )
     logging.info(
-        "Daily metric job %s OpenSearch export completed in %.3f seconds.",
+        "Daily metric job %s OpenSearch export completed in %.3f seconds; "
+        "peak RSS %.1f MiB.",
         job.pk,
         monotonic() - opensearch_started,
+        _peak_rss_mib(),
     )
+
+
+def _peak_rss_mib():
+    return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024
diff --git a/metrics/services/daily_payloads.py b/metrics/services/daily_payloads.py
index 0db8f22..132bb81 100644
--- a/metrics/services/daily_payloads.py
+++ b/metrics/services/daily_payloads.py
@@ -5,6 +5,7 @@
 from datetime import timedelta
 from pathlib import Path
 
+import ijson
 from django.conf import settings
 from django.utils import timezone
 
@@ -28,38 +29,86 @@ def resolve_storage_path(storage_path):
     return get_daily_payload_root() / storage_path
 
 
-def write_payload(storage_path, payload):
+class DailyPayloadWriter:
+    def __init__(self, storage_path, collection, access_date):
+        self.collection = collection
+        self.access_date = access_date
+        self.resolved_path = resolve_storage_path(storage_path)
+        self.tmp_path = self.resolved_path.with_suffix(
+            f"{self.resolved_path.suffix}.tmp"
+        )
+        self.encoder = json.JSONEncoder(
+            ensure_ascii=True,
+            sort_keys=True,
+            separators=(",", ":"),
+        )
+        self.payload_hash = hashlib.sha256()
+        self.output = None
+        self.next_granularity = "month"
+        self.completed = False
+
+    def __enter__(self):
+        self.resolved_path.parent.mkdir(parents=True, exist_ok=True)
+        self.output = self.tmp_path.open("wb")
+        self._write_text('{"access_date":')
+        self._write_json(self.access_date)
+        self._write_text(',"collection":')
+        self._write_json(self.collection)
+        self._write_text(',"documents":{"month":')
+        return self
+
+    def write_documents(self, granularity, documents):
+        if granularity != self.next_granularity:
+            raise RuntimeError(
+                f"Expected {self.next_granularity} documents, got {granularity}."
+            )
+
+        self._write_json(documents)
+        if granularity == "month":
+            self._write_text(',"year":')
+            self.next_granularity = "year"
+        else:
+            self.next_granularity = None
+
+    def finalize(self, input_log_hashes, summary):
+        if self.next_granularity is not None:
+            raise RuntimeError("Month and year documents must be written first.")
+
+        self._write_text('},"input_log_hashes":')
+        self._write_json(input_log_hashes)
+        self._write_text(',"summary":')
+        self._write_json(summary)
+        self._write_text("}")
+        self.output.close()
+        self.output = None
+        self.tmp_path.replace(self.resolved_path)
+        self.completed = True
+        return self.payload_hash.hexdigest()
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        if self.output is not None:
+            self.output.close()
+            self.output = None
+        if not self.completed:
+            try:
+                self.tmp_path.unlink()
+            except FileNotFoundError:
+                pass
+
+    def _write_json(self, value):
+        for chunk in self.encoder.iterencode(value):
+            self._write_text(chunk)
+
+    def _write_text(self, value):
+        encoded_value = value.encode("utf-8")
+        self.payload_hash.update(encoded_value)
+        self.output.write(encoded_value)
+
+
+def iter_document_items(storage_path, granularity):
     resolved_path = resolve_storage_path(storage_path)
-    resolved_path.parent.mkdir(parents=True, exist_ok=True)
-
-    encoder = json.JSONEncoder(
-        ensure_ascii=True,
-        sort_keys=True,
-        separators=(",", ":"),
-    )
-    payload_hash = hashlib.sha256()
-    tmp_path = resolved_path.with_suffix(f"{resolved_path.suffix}.tmp")
-
-    try:
-        with tmp_path.open("wb") as output:
-            for chunk in encoder.iterencode(payload):
-                encoded_chunk = chunk.encode("utf-8")
-                payload_hash.update(encoded_chunk)
-                output.write(encoded_chunk)
-        tmp_path.replace(resolved_path)
-    except Exception:
-        try:
-            tmp_path.unlink()
-        except FileNotFoundError:
-            pass
-        raise
-
-    return payload_hash.hexdigest()
-
-
-def read_payload(storage_path):
-    resolved_path = resolve_storage_path(storage_path)
-    return json.loads(resolved_path.read_text(encoding="utf-8"))
+    with resolved_path.open("rb") as payload_file:
+        yield from ijson.kvitems(payload_file, f"documents.{granularity}")
 
 
 def cleanup_exported_payloads(collections=None, older_than_days=7):
diff --git a/metrics/services/export.py b/metrics/services/export.py
index 4c3def9..3fa5b90 100644
--- a/metrics/services/export.py
+++ b/metrics/services/export.py
@@ -1,4 +1,7 @@
 import logging
+import resource
+from itertools import chain
+from time import monotonic
 
 from django.conf import settings
 
@@ -7,90 +10,83 @@
 from metrics.services import daily_payloads
 
 
-def load_daily_metric_payload(job):
+def daily_metric_payload_exists(job):
     if not job.storage_path:
-        return None
-    try:
-        return daily_payloads.read_payload(job.storage_path)
-    except FileNotFoundError:
+        return False
+    if not daily_payloads.resolve_storage_path(job.storage_path).is_file():
         logging.warning("Daily metric payload not found for job %s.", job.pk)
-        return None
+        return False
+    return True
 
 
-def export_daily_metric_payload(search_client, job, payload):
+def export_daily_metric_payload(search_client, job):
     if not job.job_id:
         raise RuntimeError("Daily metric job has no payload hash.")
-
-    export_documents(
-        search_client=search_client,
-        documents=payload.get("documents") or {},
-        collection=payload.get("collection") or job.collection.acron3,
-        job_id=job.job_id,
-    )
-
-
-def export_documents(search_client, documents, collection, job_id):
-    if not documents:
-        return
-
-    _sync_documents_group(
-        search_client=search_client,
-        collection=collection,
-        documents=documents.get("month", {}),
-        granularity="month",
-        job_id=job_id,
-    )
-    _sync_documents_group(
-        search_client=search_client,
-        collection=collection,
-        documents=documents.get("year", {}),
-        granularity="year",
-        job_id=job_id,
-    )
+    if not daily_metric_payload_exists(job):
+        raise RuntimeError(f"Daily metric payload not found for job {job.pk}.")
+
+    for granularity in ("month", "year"):
+        started = monotonic()
+        exported = _sync_documents_group(
+            search_client=search_client,
+            collection=job.collection.acron3,
+            access_date=job.access_date,
+            document_items=daily_payloads.iter_document_items(
+                job.storage_path,
+                granularity,
+            ),
+            granularity=granularity,
+            job_id=job.job_id,
+        )
+        logging.info(
+            "Daily metric job %s %s OpenSearch export completed in %.3f "
+            "seconds; %s documents; peak RSS %.1f MiB.",
+            job.pk,
+            granularity,
+            monotonic() - started,
+            exported,
+            _peak_rss_mib(),
+        )
 
 
 def _sync_documents_group(
     search_client,
     collection,
-    documents,
+    access_date,
+    document_items,
     granularity,
     job_id,
 ):
-    if not documents:
-        return
+    try:
+        first_item = next(document_items)
+    except StopIteration:
+        return 0
 
-    grouped_documents = {}
     index_prefix = settings.OPENSEARCH_INDEX_NAME
+    index_date = access_date.isoformat()
+    if granularity == "month":
+        index_name = generate_month_index_name(
+            index_prefix=index_prefix,
+            collection=collection,
+            date=index_date,
+        )
+    else:
+        index_name = generate_year_index_name(
+            index_prefix=index_prefix,
+            collection=collection,
+            date=index_date,
+        )
 
-    for doc_id, document in documents.items():
-        access = document.get("access") or {}
-        if granularity == "month":
-            index_name = generate_month_index_name(
-                index_prefix=index_prefix,
-                collection=collection,
-                date=f"{access.get('month')}-01",
-            )
-            mappings = get_index_mappings(collection, "month")
-        else:
-            index_name = generate_year_index_name(
-                index_prefix=index_prefix,
-                collection=collection,
-                date=f"{access.get('year')}-01-01",
-            )
-            mappings = get_index_mappings(collection, "year")
+    search_client.create_index_if_not_exists(
+        index_name=index_name,
+        mappings=get_index_mappings(collection, granularity),
+    )
+    return search_client.increment_document_items_for_daily_job(
+        index_name=index_name,
+        document_items=chain((first_item,), document_items),
+        job_id=job_id,
+    )
 
-        grouped_documents.setdefault(
-            index_name, {"mappings": mappings, "documents": {}}
-        )
-        grouped_documents[index_name]["documents"][doc_id] = document
 
-    for index_name, payload in grouped_documents.items():
-        search_client.create_index_if_not_exists(
-            index_name=index_name,
-            mappings=payload["mappings"],
-        )
-        search_client.increment_documents_for_daily_job(
-            index_name=index_name,
-            documents=payload["documents"],
-            job_id=job_id,
-        )
+def _peak_rss_mib():
+    return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024
diff --git a/metrics/services/parsing/job_payloads.py b/metrics/services/parsing/job_payloads.py
index e3dc0ba..7fb1ebf 100644
--- a/metrics/services/parsing/job_payloads.py
+++ b/metrics/services/parsing/job_payloads.py
@@ -1,4 +1,6 @@
+import gc
 import logging
+import resource
 from time import monotonic
 
 from django.conf import settings
@@ -38,20 +40,15 @@ def build_daily_metric_job_payload(job, robots_list, mmdb, track_errors=False):
         )
         _merge_log_summary(summary, log_summary)
     logging.info(
-        "Daily metric job %s parsing completed in %.3f seconds.",
+        "Daily metric job %s parsing completed in %.3f seconds; "
+        "%s compact records; peak RSS %.1f MiB.",
         job.pk,
         monotonic() - parsing_started,
+        len(results),
+        _peak_rss_mib(),
     )
 
-    conversion_started = monotonic()
-    documents = index_docs.convert(results)
-    logging.info(
-        "Daily metric job %s conversion completed in %.3f seconds.",
-        job.pk,
-        monotonic() - conversion_started,
-    )
-    payload = _write_job_payload(job, documents, summary)
-    return payload
+    return _write_job_payload(job, results, summary)
 
 
 def _get_job_log_files(job, input_log_hashes):
@@ -138,24 +135,72 @@ def _merge_log_summary(summary, log_summary):
     summary["discarded_lines"] += log_summary["discarded_lines"]
 
 
-def _write_job_payload(job, documents, summary):
-    serialization_started = monotonic()
+def _write_job_payload(job, results, summary):
     storage_path = daily_payloads.build_daily_storage_path(
         job.collection,
         job.access_date,
     )
-    payload = {
-        "collection": job.collection.acron3,
-        "access_date": job.access_date.isoformat(),
-        "input_log_hashes": summary["input_log_hashes"],
-        "documents": documents,
-        "summary": summary,
-    }
-    payload_hash = daily_payloads.write_payload(storage_path, payload)
+    month_document_count = 0
+    year_document_count = 0
+    serialization_seconds = 0
+
+    with daily_payloads.DailyPayloadWriter(
+        storage_path=storage_path,
+        collection=job.collection.acron3,
+        access_date=job.access_date.isoformat(),
+    ) as writer:
+        month_started = monotonic()
+        month_documents = index_docs.convert_granularity(
+            results.iter_materialized_values(),
+            "month",
+        )
+        month_document_count = len(month_documents)
+        month_conversion_seconds = monotonic() - month_started
+
+        serialization_started = monotonic()
+        writer.write_documents("month", month_documents)
+        serialization_seconds += monotonic() - serialization_started
+        del month_documents
+        gc.collect()
+        logging.info(
+            "Daily metric job %s monthly conversion completed in %.3f seconds; "
+            "%s documents; peak RSS %.1f MiB.",
+            job.pk,
+            month_conversion_seconds,
+            month_document_count,
+            _peak_rss_mib(),
+        )
+
+        year_started = monotonic()
+        year_documents = index_docs.convert_granularity(
+            results.iter_materialized_values(consume=True),
+            "year",
+        )
+        year_document_count = len(year_documents)
+        year_conversion_seconds = monotonic() - year_started
+        del results
+
+        serialization_started = monotonic()
+        writer.write_documents("year", year_documents)
+        del year_documents
+        payload_hash = writer.finalize(summary["input_log_hashes"], summary)
+        serialization_seconds += monotonic() - serialization_started
+        gc.collect()
+        logging.info(
+            "Daily metric job %s yearly conversion completed in %.3f seconds; "
+            "%s documents; peak RSS %.1f MiB.",
+            job.pk,
+            year_conversion_seconds,
+            year_document_count,
+            _peak_rss_mib(),
+        )
+
     logging.info(
-        "Daily metric job %s serialization completed in %.3f seconds.",
+        "Daily metric job %s serialization completed in %.3f seconds; "
+        "peak RSS %.1f MiB.",
         job.pk,
-        monotonic() - serialization_started,
+        serialization_seconds,
+        _peak_rss_mib(),
     )
 
     job.input_log_hashes = summary["input_log_hashes"]
@@ -163,8 +208,8 @@ def _write_job_payload(job, documents, summary):
     job.payload_hash = payload_hash
     job.summary = {
         **summary,
-        "month_document_count": len(documents.get("month", {})),
-        "year_document_count": len(documents.get("year", {})),
+        "month_document_count": month_document_count,
+        "year_document_count": year_document_count,
     }
     job.save(
         update_fields=[
@@ -175,4 +220,8 @@ def _write_job_payload(job, documents, summary):
             "updated",
         ]
     )
-    return payload
+    return storage_path.as_posix(), payload_hash
+
+
+def _peak_rss_mib():
+    return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss / 1024
diff --git a/metrics/tests/counter/access/test_accumulation.py b/metrics/tests/counter/access/test_accumulation.py
index 1a912ce..8f5366c 100644
--- a/metrics/tests/counter/access/test_accumulation.py
+++ b/metrics/tests/counter/access/test_accumulation.py
@@ -50,11 +50,11 @@ def _line(self, **overrides):
         return base
 
     def test_stores_source_and_periods(self):
-        results = {}
+        results = DailyAccessAccumulator()
         accumulation.accumulate(results, self._book_counter_access(), self._line())
 
         self.assertEqual(len(results), 1)
-        result = next(iter(results.values()))
+        result = next(results.iter_materialized_values())
         self.assertEqual(result["source"]["source_type"], "book")
         self.assertEqual(result["source"]["source_id"], "q7gtd")
         self.assertEqual(result["source"]["main_title"], "Book Title")
@@ -68,17 +68,17 @@ def test_stores_source_and_periods(self):
         self.assertIn("user_session_id", result)
 
     def test_rejects_invalid_local_datetime(self):
-        results = {}
+        results = DailyAccessAccumulator()
         with self.assertRaises(ValueError):
             accumulation.accumulate(
                 results,
                 self._book_counter_access(),
                 self._line(local_datetime=None),
             )
-        self.assertEqual(results, {})
+        self.assertEqual(len(results), 0)
 
     def test_does_not_expand_book_into_segments(self):
-        results = {}
+        results = DailyAccessAccumulator()
         counter_access = self._book_counter_access(
             source_id="c2248",
             pid_generic="BOOK:C2248",
@@ -95,11 +95,11 @@ def test_does_not_expand_book_into_segments(self):
         )
         accumulation.accumulate(results, counter_access, self._line())
         self.assertEqual(len(results), 1)
-        result = list(results.values())[0]
+        result = list(results.iter_materialized_values())[0]
         self.assertEqual(result["pid_generic"], "BOOK:C2248")
 
     def test_double_click_filter_uses_url_bucket_for_same_item(self):
-        results = {}
+        results = DailyAccessAccumulator()
         counter_access = self._book_counter_access(
             source_id="c2248",
             pid_generic="BOOK:C2248/CHAPTER:03",
@@ -126,14 +126,14 @@ def test_double_click_filter_uses_url_bucket_for_same_item(self):
             ),
         )
 
-        raw = next(iter(results.values()))
+        raw = next(results.iter_materialized_values())
         self.assertEqual(
             set(raw["click_timestamps_by_url"]),
             {"/id/c2248/03", "/id/c2248/epub/03.html"},
         )
 
     def test_same_url_within_window_produces_single_url_bucket(self):
-        results = {}
+        results = DailyAccessAccumulator()
         counter_access = self._book_counter_access(
             source_id="c2248",
             pid_generic="BOOK:C2248/CHAPTER:03",
@@ -160,14 +160,14 @@ def test_same_url_within_window_produces_single_url_bucket(self):
             ),
         )
 
-        raw = next(iter(results.values()))
+        raw = next(results.iter_materialized_values())
         self.assertEqual(
             raw["click_timestamps_by_url"],
             {"/id/c2248/03": {5: 1, 20: 1}},
         )
 
     def test_parses_datetime_string_and_stores_integer_seconds(self):
-        results = {}
+        results = DailyAccessAccumulator()
 
         accumulation.accumulate(
             results,
@@ -175,7 +175,7 @@ def test_parses_datetime_string_and_stores_integer_seconds(self):
             self._line(local_datetime="2024-01-15 10:01:05"),
         )
 
-        raw = next(iter(results.values()))
+        raw = next(results.iter_materialized_values())
         self.assertNotIn("click_timestamps", raw)
         self.assertEqual(
             raw["click_timestamps_by_url"],
@@ -183,25 +183,22 @@ def test_parses_datetime_string_and_stores_integer_seconds(self):
         )
 
     def test_generates_session_id_from_client_ip_datetime(self):
-        results = {}
+        results = DailyAccessAccumulator()
         accumulation.accumulate(results, self._book_counter_access(), self._line())
-        result = next(iter(results.values()))
-        self.assertEqual(
-            result["user_session_id"], "browser|1.0|127.0.0.1|2024-01-15|10"
-        )
+        result = next(results.iter_materialized_values())
+        self.assertEqual(result["user_session_id"], 1)
 
     def test_ipv6_address_is_accepted(self):
-        results = {}
+        results = DailyAccessAccumulator()
         accumulation.accumulate(
             results,
             self._book_counter_access(),
             self._line(ip_address="2001:4860:7:1103::"),
         )
-        result = next(iter(results.values()))
-        self.assertIn("2001:4860:7:1103::", result["user_session_id"])
+        result = next(results.iter_materialized_values())
+        self.assertEqual(result["user_session_id"], 1)
 
-    def test_compact_accumulator_preserves_metrics(self):
-        regular = {}
+    def test_compact_accumulator_preserves_metrics_after_repeated_events(self):
         compact = DailyAccessAccumulator()
         events = [
             self._line(
@@ -223,10 +220,13 @@ def test_compact_accumulator_preserves_metrics(self):
         ]
 
         for event in events:
-            accumulation.accumulate(regular, self._book_counter_access(), event)
             accumulation.accumulate(compact, self._book_counter_access(), event)
 
-        self.assertEqual(index_docs.convert(compact), index_docs.convert(regular))
+        values = list(compact.iter_materialized_values())
+        month = index_docs.convert_granularity(iter(values), "month")
+        year = index_docs.convert_granularity(iter(values), "year")
+        self.assertEqual(len(month), 2)
+        self.assertEqual(len(year), 2)
 
     def test_compact_accumulator_promotes_only_repeated_timestamps(self):
         compact = DailyAccessAccumulator()
@@ -240,7 +240,7 @@ def test_compact_accumulator_promotes_only_repeated_timestamps(self):
                 url="/id/q7gtd/full-text",
             ),
         )
-        record = next(iter(dict.values(compact)))
+        record = next(iter(compact._records.values()))
         self.assertIsNone(record.multiple_timestamps)
 
         accumulation.accumulate(
diff --git a/metrics/tests/counter/access/test_daily_accumulator.py b/metrics/tests/counter/access/test_daily_accumulator.py
index c02c84d..4505b5f 100644
--- a/metrics/tests/counter/access/test_daily_accumulator.py
+++ b/metrics/tests/counter/access/test_daily_accumulator.py
@@ -1,68 +1,64 @@
 from metrics.counter.access.daily_accumulator import DailyAccessAccumulator
 
 
-def _record(session_id, **overrides):
-    record = {
+def _record(pid_v3):
+    return {
         "collection": "scl",
         "source_key": "1234-5678",
         "document_type": "article",
         "pid_v2": "S123456782026000100001",
-        "pid_v3": "abc123",
+        "pid_v3": pid_v3,
         "pid_generic": None,
         "title_pid_generic": None,
         "source": {"source_id": "1234-5678", "main_title": "Journal"},
-        "document": {"title": "Article"},
-        "user_session_id": session_id,
+        "document": {"title": f"Article {pid_v3}"},
     }
-    record.update(overrides)
-    return record
 
 
-def test_interns_repeated_metadata_and_sessions():
-    accumulator = DailyAccessAccumulator()
-    first_session = "|".join(["Firefox", "1", "127.0.0.1", "2026-08-25", "10"])
-    second_session = "|".join(["Firefox", "1", "127.0.0.1", "2026-08-25", "10"])
-    assert first_session is not second_session
-
-    accumulator["first"] = _record(first_session)
-    accumulator["second"] = _record(second_session)
-
-    assert accumulator["first"]["source"] is accumulator["second"]["source"]
-    assert accumulator["first"]["document"] is accumulator["second"]["document"]
-    assert (
-        accumulator["first"]["user_session_id"]
-        is accumulator["second"]["user_session_id"]
+def _accumulate(accumulator, pid_v3, session_ip):
+    accumulator.accumulate_access(
+        data=_record(pid_v3),
+        session_key=("Firefox", "1", session_ip, 739491, 10),
+        url=f"/{pid_v3}",
+        second=5,
     )
 
 
-def test_interns_empty_document_metadata_for_the_same_document():
+def test_materialization_preserves_insertion_order():
     accumulator = DailyAccessAccumulator()
+    _accumulate(accumulator, "first", "127.0.0.1")
+    _accumulate(accumulator, "second", "127.0.0.2")
 
-    accumulator["first"] = _record("first", document={})
-    accumulator["second"] = _record("second", document={})
+    values = list(accumulator.iter_materialized_values())
 
-    assert accumulator["first"]["document"] is accumulator["second"]["document"]
+    assert [value["pid_v3"] for value in values] == ["first", "second"]
 
 
-def test_does_not_share_metadata_between_distinct_documents():
+def test_consuming_materialization_releases_all_internal_structures():
     accumulator = DailyAccessAccumulator()
+    _accumulate(accumulator, "first", "127.0.0.1")
+    _accumulate(accumulator, "second", "127.0.0.2")
 
-    accumulator["first"] = _record("first", pid_v3="first", document={})
-    accumulator["second"] = _record("second", pid_v3="second", document={})
+    values = list(accumulator.iter_materialized_values(consume=True))
 
-    assert accumulator["first"]["document"] is not accumulator["second"]["document"]
+    assert [value["pid_v3"] for value in values] == ["first", "second"]
+    assert len(accumulator) == 0
+    assert accumulator._documents == []
+    assert accumulator._document_ids == {}
+    assert accumulator._sources == []
+    assert accumulator._source_ids == {}
+    assert accumulator._sessions == {}
+    assert accumulator._strings == []
+    assert accumulator._string_ids == {}
 
 
-def test_does_not_share_documents_without_identifiers():
+def test_consuming_materialization_releases_structures_after_consumer_error():
     accumulator = DailyAccessAccumulator()
-    identifiers = {
-        "pid_v2": None,
-        "pid_v3": None,
-        "pid_generic": None,
-        "title_pid_generic": None,
-    }
+    _accumulate(accumulator, "first", "127.0.0.1")
+    values = accumulator.iter_materialized_values(consume=True)
 
-    accumulator["first"] = _record("first", document={}, **identifiers)
-    accumulator["second"] = _record("second", document={}, **identifiers)
+    next(values)
+    values.close()
 
-    assert accumulator["first"]["document"] is not accumulator["second"]["document"]
+    assert len(accumulator) == 0
+    assert accumulator._documents == []
diff --git a/metrics/tests/counter/indexing/test_converter.py b/metrics/tests/counter/indexing/test_converter.py
index 184e871..1bc3888 100644
--- a/metrics/tests/counter/indexing/test_converter.py
+++ b/metrics/tests/counter/indexing/test_converter.py
@@ -10,6 +10,14 @@
 from metrics.counter.indexing import converter as index_docs
 
 
+def _convert(data):
+    values = list(data.values())
+    return {
+        "month": index_docs.convert_granularity(iter(values), "month"),
+        "year": index_docs.convert_granularity(iter(values), "year"),
+    }
+
+
 class TestConverter(unittest.TestCase):
     def test_creates_month_and_year_views_for_book_chapter(self):
         data = {
@@ -47,7 +55,7 @@ def test_creates_month_and_year_views_for_book_chapter(self):
             }
         }
 
-        metrics_data = index_docs.convert(data)
+        metrics_data = _convert(data)
 
         self.assertEqual(set(metrics_data.keys()), {"month", "year"})
         self.assertEqual(len(metrics_data["month"]), 2)
@@ -156,7 +164,7 @@ def test_maps_counter_data_types_for_preprint_and_dataset(self):
             },
         }
 
-        metrics_data = index_docs.convert(data)
+        metrics_data = _convert(data)
         preprint_doc = metrics_data["month"][
             "preprints|scielo-preprints|||10.1590/SCIELOPREPRINTS.1234|2024-01|Open|Regular|2024"
         ]
@@ -225,7 +233,7 @@ def test_dedupes_book_unique_item_across_formats(self):
             },
         }
 
-        metrics_data = index_docs.convert(data)
+        metrics_data = _convert(data)
         month_item = metrics_data["month"][
             "books|c2248|||BOOK:C2248/CHAPTER:03|2024-01|Open|Regular|2018"
         ]
@@ -270,7 +278,7 @@ def test_skips_book_landing_page_from_item_scope(self):
             },
         }
 
-        metrics_data = index_docs.convert(data)
+        metrics_data = _convert(data)
         self.assertEqual(
             set(metrics_data["month"].keys()),
             {"title|books|c2248|||BOOK:C2248|2024-01|Open|Regular|2018"},
@@ -310,7 +318,7 @@ def test_whole_book_without_segments_counts_as_book_segment(self):
             },
         }
 
-        metrics_data = index_docs.convert(data)
+        metrics_data = _convert(data)
         month_item = metrics_data["month"][
             "books|c2248|||BOOK:C2248|2024-01|Open|Regular|2018"
         ]
@@ -371,7 +379,7 @@ def test_aggregates_multiple_chapters_at_title_level(self):
             },
         }
 
-        metrics_data = index_docs.convert(data)
+        metrics_data = _convert(data)
         self.assertEqual(len(metrics_data["month"]), 3)
         self.assertEqual(len(metrics_data["year"]), 3)
 
@@ -387,8 +395,9 @@ def test_double_click_collapses_same_url_within_30_seconds(self):
         from datetime import datetime
 
         from metrics.counter.access import accumulation
+        from metrics.counter.access.daily_accumulator import DailyAccessAccumulator
 
-        results = {}
+        results = DailyAccessAccumulator()
         counter_access = {
             "collection": "books",
             "source_type": "book",
@@ -423,7 +432,11 @@ def test_double_click_collapses_same_url_within_30_seconds(self):
             {**base_line, "local_datetime": datetime(2024, 1, 15, 10, 0, 20)},
         )
 
-        metrics_data = index_docs.convert(results)
+        values = list(results.iter_materialized_values())
+        metrics_data = {
+            "month": index_docs.convert_granularity(iter(values), "month"),
+            "year": index_docs.convert_granularity(iter(values), "year"),
+        }
         month_item = metrics_data["month"][
             "books|c2248|||BOOK:C2248/CHAPTER:03|2024-01|Open|Regular|2018"
         ]
@@ -457,7 +470,7 @@ def test_article_pipeline_sets_journal_parent(self):
             }
         }
 
-        metrics_data = index_docs.convert(data)
+        metrics_data = _convert(data)
         month_doc = list(metrics_data["month"].values())[0]
 
         self.assertEqual(month_doc["counter"]["data_type"], "Article")
@@ -467,6 +480,5 @@ def test_article_pipeline_sets_journal_parent(self):
         self.assertEqual(month_doc["total_requests"], 1)
         self.assertEqual(month_doc["total_investigations"], 1)
 
-    def test_non_dict_input_returns_empty(self):
-        result = index_docs.convert(None)
-        self.assertEqual(result, {"month": {}, "year": {}})
+    def test_empty_iterable_returns_empty(self):
+        self.assertEqual(index_docs.convert_granularity(iter(()), "month"), {})
diff --git a/metrics/tests/helpers.py b/metrics/tests/helpers.py
new file mode 100644
index 0000000..0d6a47c
--- /dev/null
+++ b/metrics/tests/helpers.py
@@ -0,0 +1,9 @@
+from metrics.counter.indexing import converter
+
+
+def convert_accumulator(accumulator):
+    values = list(accumulator.iter_materialized_values())
+    return {
+        "month": converter.convert_granularity(iter(values), "month"),
+        "year": converter.convert_granularity(iter(values), "year"),
+    }
diff --git a/metrics/tests/integration/test_books_log_to_metrics.py b/metrics/tests/integration/test_books_log_to_metrics.py
index b898a9b..6edf141 100644
--- a/metrics/tests/integration/test_books_log_to_metrics.py
+++ b/metrics/tests/integration/test_books_log_to_metrics.py
@@ -2,12 +2,13 @@
 from datetime import datetime
 from pathlib import Path
 
+from scielo_usage_counter import log_handler
 from scielo_usage_counter.translator.books import URLTranslatorBooksSite
 from scielo_usage_counter.url_translator import URLTranslationManager
 
 from metrics.counter.access import accumulation, extraction, validation
-from metrics.counter.indexing import converter as index_docs
-from scielo_usage_counter import log_handler
+from metrics.counter.access.daily_accumulator import DailyAccessAccumulator
+from metrics.tests.helpers import convert_accumulator
 
 FIXTURES_DIR = Path(__file__).resolve().parent.parent / "fixtures"
 
@@ -85,7 +86,7 @@ def test_pdf_and_epub_formats_detected(self):
         self.assertTrue(len(formats) > 0)
 
     def test_full_pipeline_with_synthetic_metadata(self):
-        results = {}
+        results = DailyAccessAccumulator()
         counter_access = extraction.extract(
             "books",
             {
@@ -117,7 +118,7 @@ def test_full_pipeline_with_synthetic_metadata(self):
             },
         )
 
-        metrics = index_docs.convert(results)
+        metrics = convert_accumulator(results)
         self.assertGreater(len(metrics["month"]), 0)
         self.assertGreater(len(metrics["year"]), 0)
 
@@ -136,7 +137,7 @@ def test_full_pipeline_with_synthetic_metadata(self):
         self.assertTrue(has_title)
 
     def test_all_metric_fields_present_in_converted_document(self):
-        results = {}
+        results = DailyAccessAccumulator()
         counter_access = extraction.extract(
             "books",
             {
@@ -166,7 +167,7 @@ def test_all_metric_fields_present_in_converted_document(self):
             },
         )
 
-        metrics = index_docs.convert(results)
+        metrics = convert_accumulator(results)
         for doc in metrics["month"].values():
             self.assertIn("total_requests", doc)
             self.assertIn("total_investigations", doc)
diff --git a/metrics/tests/integration/test_bunnynet_log_to_metrics.py b/metrics/tests/integration/test_bunnynet_log_to_metrics.py
index 434fbaa..2865bc7 100644
--- a/metrics/tests/integration/test_bunnynet_log_to_metrics.py
+++ b/metrics/tests/integration/test_bunnynet_log_to_metrics.py
@@ -5,7 +5,8 @@
 from scielo_usage_counter.values import CONTENT_TYPE_FULL_TEXT, MEDIA_FORMAT_HTML
 
 from metrics.counter.access import accumulation, extraction
-from metrics.counter.indexing import converter as index_docs
+from metrics.counter.access.daily_accumulator import DailyAccessAccumulator
+from metrics.tests.helpers import convert_accumulator
 
 FIXTURES_DIR = Path(__file__).resolve().parent.parent / "fixtures"
 
@@ -72,10 +73,10 @@ def test_uses_client_country_and_millisecond_timestamp_in_year_metrics(self):
                 "media_language": "en",
             },
         )
-        results = {}
+        results = DailyAccessAccumulator()
 
         accumulation.accumulate(results, counter_access, line)
-        metrics = index_docs.convert(results)
+        metrics = convert_accumulator(results)
 
         self.assertEqual(line["country_code"], "US")
         self.assertEqual(line["local_datetime"], "2026-08-04 23:59:59")
diff --git a/metrics/tests/integration/test_classic_log_to_metrics.py b/metrics/tests/integration/test_classic_log_to_metrics.py
index 6480bc7..e75b79e 100644
--- a/metrics/tests/integration/test_classic_log_to_metrics.py
+++ b/metrics/tests/integration/test_classic_log_to_metrics.py
@@ -1,12 +1,13 @@
 import unittest
 from pathlib import Path
 
+from scielo_usage_counter import log_handler
 from scielo_usage_counter.translator.classic import URLTranslatorClassicSite
 from scielo_usage_counter.url_translator import URLTranslationManager
 
 from metrics.counter.access import accumulation, extraction, validation
-from metrics.counter.indexing import converter as index_docs
-from scielo_usage_counter import log_handler
+from metrics.counter.access.daily_accumulator import DailyAccessAccumulator
+from metrics.tests.helpers import convert_accumulator
 
 FIXTURES_DIR = Path(__file__).resolve().parent.parent / "fixtures"
 
@@ -35,7 +36,7 @@ def _parse_log(self):
 
     def _full_pipeline(self):
         lines, stats = self._parse_log()
-        results = {}
+        results = DailyAccessAccumulator()
         valid_count = 0
 
         for line in lines:
@@ -75,7 +76,7 @@ def test_produces_article_type_metrics(self):
             self.skipTest("No valid lines in classic fixture for this translator")
             return
 
-        metrics = index_docs.convert(results)
+        metrics = convert_accumulator(results)
 
         for doc in metrics["month"].values():
             self.assertEqual(doc["counter"]["data_type"], "Article")
@@ -88,7 +89,7 @@ def test_sets_journal_parent_data_type(self):
             self.skipTest("No valid lines")
             return
 
-        metrics = index_docs.convert(results)
+        metrics = convert_accumulator(results)
         for doc in metrics["month"].values():
             source_type = doc.get("source", {}).get("type")
             if source_type == "journal":
@@ -100,6 +101,6 @@ def test_handles_truncated_user_agent(self):
 
     def test_valid_lines_produce_session_ids(self):
         results, _, _, _ = self._full_pipeline()
-        for value in results.values():
+        for value in results.iter_materialized_values():
             self.assertIn("user_session_id", value)
             self.assertIsNotNone(value["user_session_id"])
diff --git a/metrics/tests/integration/test_pipelines.py b/metrics/tests/integration/test_pipelines.py
index 95e700f..292bab2 100644
--- a/metrics/tests/integration/test_pipelines.py
+++ b/metrics/tests/integration/test_pipelines.py
@@ -8,7 +8,8 @@
 )
 
 from metrics.counter.access import accumulation, extraction
-from metrics.counter.indexing import converter as index_docs
+from metrics.counter.access.daily_accumulator import DailyAccessAccumulator
+from metrics.tests.helpers import convert_accumulator
 
 
 class TestPreprintPipeline(unittest.TestCase):
@@ -30,7 +31,7 @@ def test_extraction_sets_preprint_types(self):
 
     def test_full_pipeline_produces_preprint_article_version(self):
         counter_access = self._build_preprint_access()
-        results = {}
+        results = DailyAccessAccumulator()
         line = {
             "client_name": "browser",
             "client_version": "1.0",
@@ -39,7 +40,7 @@ def test_full_pipeline_produces_preprint_article_version(self):
             "local_datetime": datetime(2024, 6, 15, 14, 30, 10),
         }
         accumulation.accumulate(results, counter_access, line)
-        metrics = index_docs.convert(results)
+        metrics = convert_accumulator(results)
 
         month_docs = list(metrics["month"].values())
         self.assertEqual(len(month_docs), 1)
@@ -71,7 +72,7 @@ def test_extraction_sets_dataset_types(self):
 
     def test_full_pipeline_produces_dataset_metrics(self):
         counter_access = self._build_dataset_access()
-        results = {}
+        results = DailyAccessAccumulator()
         line = {
             "client_name": "browser",
             "client_version": "1.0",
@@ -80,7 +81,7 @@ def test_full_pipeline_produces_dataset_metrics(self):
             "local_datetime": datetime(2024, 6, 15, 14, 30, 10),
         }
         accumulation.accumulate(results, counter_access, line)
-        metrics = index_docs.convert(results)
+        metrics = convert_accumulator(results)
 
         month_docs = list(metrics["month"].values())
         self.assertEqual(len(month_docs), 1)
@@ -109,7 +110,7 @@ def test_opac_article_produces_journal_article_metrics(self):
             },
         )
 
-        results = {}
+        results = DailyAccessAccumulator()
         line = {
             "client_name": "Chrome",
             "client_version": "120.0",
@@ -118,7 +119,7 @@ def test_opac_article_produces_journal_article_metrics(self):
             "local_datetime": datetime(2024, 3, 20, 8, 15, 42),
         }
         accumulation.accumulate(results, counter_access, line)
-        metrics = index_docs.convert(results)
+        metrics = convert_accumulator(results)
 
         doc = list(metrics["month"].values())[0]
         self.assertEqual(doc["counter"]["data_type"], "Article")
diff --git a/metrics/tests/opensearch/test_client.py b/metrics/tests/opensearch/test_client.py
index 9eb7ebc..d3ec0db 100644
--- a/metrics/tests/opensearch/test_client.py
+++ b/metrics/tests/opensearch/test_client.py
@@ -48,6 +48,7 @@ def test_verify_certs_false_explicitly_overrides_settings(self, mock_opensearch)
         mock_opensearch.assert_called_once_with(
             "https://example.org:9200",
             verify_certs=False,
+            http_compress=True,
         )
 
     def test_get_index_mappings_returns_books_specific_mappings(self):
@@ -104,19 +105,27 @@ def test_increment_documents_for_daily_job_uses_applied_jobs(
         mock_get_client.return_value = Mock()
         client = OpenSearchUsageClient(url="https://example.org:9200")
 
-        client.increment_documents_for_daily_job(
+        mock_bulk.return_value = (1, 0)
+        documents = iter(
+            [
+                (
+                    "doc-1",
+                    {
+                        "collection": "books",
+                        "document": {"id": "BOOK:WD"},
+                        "access": {"month": "2025-06"},
+                        "total_requests": 3,
+                        "total_investigations": 4,
+                        "unique_requests": 2,
+                        "unique_investigations": 3,
+                    },
+                )
+            ]
+        )
+
+        succeeded = client.increment_document_items_for_daily_job(
             index_name="usage_monthly_books_202506",
-            documents={
-                "doc-1": {
-                    "collection": "books",
-                    "document": {"id": "BOOK:WD"},
-                    "access": {"month": "2025-06"},
-                    "total_requests": 3,
-                    "total_investigations": 4,
-                    "unique_requests": 2,
-                    "unique_investigations": 3,
-                }
-            },
+            document_items=documents,
             job_id="books|2025-06-03|abc123",
         )
 
@@ -128,3 +137,5 @@ def test_increment_documents_for_daily_job_uses_applied_jobs(
             action["script"]["params"]["job_id"], "books|2025-06-03|abc123"
         )
         self.assertEqual(action["upsert"], {"applied_jobs": []})
+        self.assertEqual(succeeded, 1)
+        self.assertEqual(mock_bulk.call_args.kwargs["chunk_size"], 500)
diff --git a/metrics/tests/parsing/test_process_line.py b/metrics/tests/parsing/test_process_line.py
index 549d93d..71586b1 100644
--- a/metrics/tests/parsing/test_process_line.py
+++ b/metrics/tests/parsing/test_process_line.py
@@ -7,6 +7,7 @@
 from collection.models import Collection
 from log_manager import choices
 from log_manager.models import LogFile
+from metrics.counter.access.daily_accumulator import DailyAccessAccumulator
 from metrics.services.parsing.lines import process_line
 
 
@@ -52,7 +53,7 @@ def _line(self, **overrides):
         return base
 
     def test_discards_invalid_local_datetime_without_raising(self):
-        results = {}
+        results = DailyAccessAccumulator()
         is_valid, error = process_line(
             results=results,
             line=self._line(),
@@ -61,10 +62,10 @@ def test_discards_invalid_local_datetime_without_raising(self):
         )
         self.assertFalse(is_valid)
         self.assertIsNone(error)
-        self.assertEqual(results, {})
+        self.assertEqual(len(results), 0)
 
     def test_url_translation_error_returns_false_none(self):
-        results = {}
+        results = DailyAccessAccumulator()
         is_valid, error = process_line(
             results=results,
             line=self._line(),
@@ -77,7 +78,7 @@ def test_url_translation_error_returns_false_none(self):
     def test_valid_line_accumulates_result(self):
         from datetime import datetime
 
-        results = {}
+        results = DailyAccessAccumulator()
         is_valid, error = process_line(
             results=results,
             line=self._line(local_datetime=datetime(2024, 1, 15, 10, 0, 5)),
@@ -89,7 +90,7 @@ def test_valid_line_accumulates_result(self):
         self.assertEqual(len(results), 1)
 
     def test_validation_failure_without_track_errors_returns_no_discarded_line(self):
-        results = {}
+        results = DailyAccessAccumulator()
         utm = self._fake_utm(
             translate_return={
                 "pid_generic": "",
@@ -108,7 +109,7 @@ def test_validation_failure_without_track_errors_returns_no_discarded_line(self)
         self.assertIsNone(error)
 
     def test_extraction_error_returns_false_none(self):
-        results = {}
+        results = DailyAccessAccumulator()
         utm = self._fake_utm(translate_return="not-a-dict")
         is_valid, error = process_line(
             results=results,
diff --git a/metrics/tests/services/test_daily_jobs.py b/metrics/tests/services/test_daily_jobs.py
index 0413ba6..0db79c4 100644
--- a/metrics/tests/services/test_daily_jobs.py
+++ b/metrics/tests/services/test_daily_jobs.py
@@ -1,3 +1,5 @@
+import tempfile
+import weakref
 from datetime import date, timedelta
 from types import SimpleNamespace
 from unittest.mock import Mock, patch
@@ -8,13 +10,21 @@
 from collection.models import Collection
 from log_manager import choices
 from log_manager.models import LogFile
+from metrics.counter.access.daily_accumulator import DailyAccessAccumulator
 from metrics.models import DailyMetricJob
 from metrics.services.jobs import (
     create_or_update_daily_metric_job,
     mark_daily_metric_job_exported,
     release_stale_daily_metric_jobs,
 )
-from metrics.services.parsing.job_payloads import build_daily_metric_job_payload
+from metrics.services.parsing.job_payloads import (
+    _write_job_payload,
+    build_daily_metric_job_payload,
+)
+
+
+class TrackableDict(dict):
+    pass
 
 
 class DailyMetricJobServiceTests(TestCase):
@@ -133,12 +143,8 @@ def test_mark_daily_metric_job_exported_sets_status_and_timestamp(self):
         self.assertIsNotNone(job.exported_at)
 
     @patch(
-        "metrics.services.parsing.job_payloads.daily_payloads.write_payload",
-        return_value="payload-hash",
-    )
-    @patch(
-        "metrics.services.parsing.job_payloads.index_docs.convert",
-        return_value={"month": {}, "year": {}},
+        "metrics.services.parsing.job_payloads.index_docs.convert_granularity",
+        side_effect=({}, {}),
     )
     @patch(
         "metrics.services.parsing.job_payloads.process_line", return_value=(True, None)
@@ -149,7 +155,6 @@ def test_build_daily_metric_job_payload_uses_only_input_log_hashes(
         mock_setup_parsing_environment,
         mock_process_line,
         mock_convert_documents,
-        mock_write_payload,
     ):
         selected = self._log_file("1" * 32)
         extra = self._log_file("2" * 32)
@@ -165,15 +170,18 @@ def test_build_daily_metric_job_payload_uses_only_input_log_hashes(
         parser.parse.return_value = [{"url": "/selected"}]
         mock_setup_parsing_environment.return_value = (parser, Mock())
 
-        payload = build_daily_metric_job_payload(
-            job, robots_list=["robot"], mmdb=Mock(data={})
-        )
+        with tempfile.TemporaryDirectory() as media_root:
+            with self.settings(MEDIA_ROOT=media_root):
+                storage_path, payload_hash = build_daily_metric_job_payload(
+                    job, robots_list=["robot"], mmdb=Mock(data={})
+                )
 
         selected.refresh_from_db()
         extra.refresh_from_db()
         job.refresh_from_db()
 
-        self.assertEqual(payload["input_log_hashes"], [selected.hash])
+        self.assertEqual(storage_path, "books/2012/03/2012-03-10.json")
+        self.assertTrue(payload_hash)
         self.assertEqual(job.input_log_hashes, [selected.hash])
         self.assertEqual(selected.status, choices.LOG_FILE_STATUS_PARSING)
         self.assertEqual(extra.status, choices.LOG_FILE_STATUS_QUEUED)
@@ -208,3 +216,47 @@ def test_build_daily_metric_job_payload_rejects_missing_input_hashes(self):
             build_daily_metric_job_payload(
                 job, robots_list=["robot"], mmdb=Mock(data={})
             )
+
+    @patch("metrics.services.parsing.job_payloads.index_docs.convert_granularity")
+    def test_payload_generation_releases_each_granularity_and_accumulator(
+        self,
+        mock_convert,
+    ):
+        job = DailyMetricJob.objects.create(
+            collection=self.collection,
+            access_date=date(2012, 3, 10),
+            status=DailyMetricJob.STATUS_EXPORTING,
+            input_log_hashes=["1" * 32],
+        )
+        accumulator = DailyAccessAccumulator()
+        accumulator.accumulate_access(
+            data={"collection": "books"},
+            session_key=("browser", "1", "127.0.0.1", 734572, 10),
+            url="/book",
+            second=5,
+        )
+        references = {}
+
+        def convert(values, granularity):
+            list(values)
+            documents = TrackableDict()
+            references[granularity] = weakref.ref(documents)
+            return documents
+
+        mock_convert.side_effect = convert
+        summary = {
+            "log_files": 1,
+            "input_log_hashes": ["1" * 32],
+            "lines_parsed": 1,
+            "valid_lines": 1,
+            "discarded_lines": 0,
+        }
+
+        with tempfile.TemporaryDirectory() as media_root:
+            with self.settings(MEDIA_ROOT=media_root):
+                _write_job_payload(job, accumulator, summary)
+
+        self.assertIsNone(references["month"]())
+        self.assertIsNone(references["year"]())
+        self.assertEqual(len(accumulator), 0)
+        self.assertEqual(accumulator._documents, [])
diff --git a/metrics/tests/services/test_daily_metric_exports.py b/metrics/tests/services/test_daily_metric_exports.py
new file mode 100644
index 0000000..c24fd88
--- /dev/null
+++ b/metrics/tests/services/test_daily_metric_exports.py
@@ -0,0 +1,118 @@
+import tempfile
+from datetime import date
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import Mock, patch
+
+from django.test import SimpleTestCase, override_settings
+
+from metrics.services import daily_payloads
+from metrics.services.daily_metric_exports import _ensure_payload
+from metrics.services.export import export_daily_metric_payload
+
+
+class DailyMetricExportTests(SimpleTestCase):
+    def setUp(self):
+        self.temporary_directory = tempfile.TemporaryDirectory()
+        self.settings_override = override_settings(
+            MEDIA_ROOT=self.temporary_directory.name,
+            OPENSEARCH_INDEX_NAME="usage",
+        )
+        self.settings_override.enable()
+        self.storage_path = Path("scl/2026/08/2026-08-25.json")
+        self.job = SimpleNamespace(
+            pk=1,
+            collection=SimpleNamespace(acron3="scl"),
+            access_date=date(2026, 8, 25),
+            storage_path=self.storage_path.as_posix(),
+            payload_hash="payload-hash",
+            job_id="scl|2026-08-25|payload-hash",
+        )
+
+    def tearDown(self):
+        self.settings_override.disable()
+        self.temporary_directory.cleanup()
+
+    def _write_payload(self):
+        with daily_payloads.DailyPayloadWriter(
+            self.storage_path,
+            "scl",
+            "2026-08-25",
+        ) as writer:
+            writer.write_documents(
+                "month",
+                {"month-doc": {"access": {"month": "2026-08"}}},
+            )
+            writer.write_documents(
+                "year",
+                {"year-doc": {"access": {"year": "2026"}}},
+            )
+            writer.finalize(["abc"], {"valid_lines": 1})
+
+    def test_export_streams_each_granularity_as_document_items(self):
+        self._write_payload()
+        search_client = Mock()
+        exported_groups = []
+
+        def consume_items(index_name, document_items, job_id):
+            exported_groups.append((index_name, list(document_items), job_id))
+            return 1
+
+        search_client.increment_document_items_for_daily_job.side_effect = consume_items
+
+        export_daily_metric_payload(search_client, self.job)
+
+        self.assertEqual(
+            exported_groups,
+            [
+                (
+                    "usage_monthly_scl_2026",
+                    [("month-doc", {"access": {"month": "2026-08"}})],
+                    self.job.job_id,
+                ),
+                (
+                    "usage_yearly_scl_2026",
+                    [("year-doc", {"access": {"year": "2026"}})],
+                    self.job.job_id,
+                ),
+            ],
+        )
+
+    def test_retry_after_partial_export_reuses_same_payload_and_job_id(self):
+        self._write_payload()
+        first_client = Mock()
+        first_client.increment_document_items_for_daily_job.side_effect = [
+            1,
+            RuntimeError("year export failed"),
+        ]
+
+        with self.assertRaisesMessage(RuntimeError, "year export failed"):
+            export_daily_metric_payload(first_client, self.job)
+
+        second_client = Mock()
+        second_client.increment_document_items_for_daily_job.side_effect = (
+            lambda index_name, document_items, job_id: len(list(document_items))
+        )
+        export_daily_metric_payload(second_client, self.job)
+
+        self.assertEqual(
+            [
+                call.kwargs["job_id"]
+                for call in second_client.increment_document_items_for_daily_job.call_args_list
+            ],
+            [self.job.job_id, self.job.job_id],
+        )
+
+    @patch("metrics.services.daily_metric_exports.fetch_required_resources")
+    @patch("metrics.services.daily_metric_exports.build_daily_metric_job_payload")
+    def test_resume_uses_persisted_payload_without_parsing(
+        self,
+        mock_build_payload,
+        mock_fetch_resources,
+    ):
+        self._write_payload()
+
+        _ensure_payload(self.job, track_errors=False, robots_source="counter")
+
+        mock_build_payload.assert_not_called()
+        mock_fetch_resources.assert_not_called()
diff --git a/metrics/tests/services/test_daily_payloads.py b/metrics/tests/services/test_daily_payloads.py
index 98dfff6..d3f169b 100644
--- a/metrics/tests/services/test_daily_payloads.py
+++ b/metrics/tests/services/test_daily_payloads.py
@@ -20,10 +20,16 @@ def tearDown(self):
         self.settings_override.disable()
         self.temporary_directory.cleanup()
 
-    def test_write_payload_preserves_canonical_bytes_and_hash(self):
+    def test_incremental_writer_preserves_canonical_bytes_and_hash(self):
+        storage_path = Path("scl/2026/08/2026-08-25.json")
         payload = {
             "collection": "scl",
-            "documents": {"á": {"total_requests": 2}},
+            "access_date": "2026-08-25",
+            "input_log_hashes": ["abc"],
+            "documents": {
+                "month": {"á": {"total_requests": 2}},
+                "year": {"z": {"total_requests": 3}},
+            },
             "summary": {"valid_lines": 1},
         }
         expected = json.dumps(
@@ -33,23 +39,66 @@ def test_write_payload_preserves_canonical_bytes_and_hash(self):
             separators=(",", ":"),
         ).encode("utf-8")
 
-        payload_hash = daily_payloads.write_payload(
-            Path("scl/2026/08/2026-08-25.json"),
-            payload,
-        )
+        with daily_payloads.DailyPayloadWriter(
+            storage_path,
+            payload["collection"],
+            payload["access_date"],
+        ) as writer:
+            writer.write_documents("month", payload["documents"]["month"])
+            writer.write_documents("year", payload["documents"]["year"])
+            payload_hash = writer.finalize(
+                payload["input_log_hashes"],
+                payload["summary"],
+            )
 
-        resolved_path = daily_payloads.resolve_storage_path(
-            Path("scl/2026/08/2026-08-25.json")
-        )
+        resolved_path = daily_payloads.resolve_storage_path(storage_path)
         self.assertEqual(resolved_path.read_bytes(), expected)
         self.assertEqual(payload_hash, hashlib.sha256(expected).hexdigest())
 
-    def test_write_payload_removes_temporary_file_after_serialization_error(self):
+    def test_iter_document_items_reads_each_granularity_incrementally(self):
         storage_path = Path("scl/2026/08/2026-08-25.json")
+        payload = {
+            "collection": "scl",
+            "access_date": "2026-08-25",
+            "input_log_hashes": ["abc"],
+            "documents": {
+                "month": {"month-1": {"total_requests": 2}},
+                "year": {"year-1": {"total_requests": 3}},
+            },
+            "summary": {},
+        }
+        with daily_payloads.DailyPayloadWriter(
+            storage_path,
+            payload["collection"],
+            payload["access_date"],
+        ) as writer:
+            writer.write_documents("month", payload["documents"]["month"])
+            writer.write_documents("year", payload["documents"]["year"])
+            writer.finalize(payload["input_log_hashes"], payload["summary"])
 
-        with self.assertRaises(TypeError):
-            daily_payloads.write_payload(storage_path, {"invalid": object()})
+        self.assertEqual(
+            list(daily_payloads.iter_document_items(storage_path, "month")),
+            [("month-1", {"total_requests": 2})],
+        )
+        self.assertEqual(
+            list(daily_payloads.iter_document_items(storage_path, "year")),
+            [("year-1", {"total_requests": 3})],
+        )
 
+    def test_incremental_writer_removes_temporary_file_after_error(self):
+        storage_path = Path("scl/2026/08/2026-08-25.json")
         resolved_path = daily_payloads.resolve_storage_path(storage_path)
-        self.assertFalse(resolved_path.exists())
+        resolved_path.parent.mkdir(parents=True)
+        resolved_path.write_bytes(b"previous canonical payload")
+
+        with self.assertRaises(TypeError):
+            with daily_payloads.DailyPayloadWriter(
+                storage_path,
+                "scl",
+                "2026-08-25",
+            ) as writer:
+                writer.write_documents("month", {})
+                writer.write_documents("year", {"invalid": object()})
+
+        self.assertEqual(resolved_path.read_bytes(), b"previous canonical payload")
         self.assertFalse(resolved_path.with_suffix(".json.tmp").exists())
diff --git a/requirements/base.txt b/requirements/base.txt
index c09ad5d..e28d18d 100644
--- a/requirements/base.txt
+++ b/requirements/base.txt
@@ -71,6 +71,9 @@ git+https://github.com/scieloorg/scielo_scholarly_data@v0.1.4#egg=scielo_scholar
 # SciELO Usage COUNTER
 git+https://github.com/scieloorg/scielo_usage_counter@2.2.1#egg=scielo_usage_counter
 
+# Incremental JSON
+ijson==3.5.1
+
 # Device Detector
 device-detector==0.10  # https://github.com/thinkwelltwd/device_detector