Hourly delivery in v2 (from Nov 3, 2026)

• The bucket, and the way you access it, with an AWS account or a Google account.
• The Parquet format and every column, tenant_id included.
• The 19 tables in the table catalog.
• Every folder already in v1/. Nothing is moved or deleted, and you can keep reading it.
• The storage rules: files still move to colder storage classes as they age, and to S3 Glacier Flexible Retrieval after 180 days.

​

What changes

​

Tables rewritten in full

s3://<YOUR_BUCKET>/v2/datamart_claudia/claudia_conversation/snapshot_date=2026-11-03/snapshot_hour=03/

​

Tables delivered in parts

s3://<YOUR_BUCKET>/v2/datamart_claudia/claudia_messages/etl_date=2026-11-03/etl_hour=03/

• In v1/, etl_date=__initial__ and the daily etl_date= folders. They no longer change after November 2, so you need to read them only once.
• In v2/, the etl_date= / etl_hour= folders, as they arrive.

​

One manifest per hour

s3://<YOUR_BUCKET>/v2/_manifests/snapshot_date=<DATE>/snapshot_hour=<HOUR>/manifest.json

​

When it arrives

​

The switch

​

What to review

​

Prepare the switch ahead of time

import boto3
from awsglue.context import GlueContext
from pyspark.context import SparkContext
from pyspark.sql import Window
from pyspark.sql import functions as F

BUCKET = "<YOUR_BUCKET>"

s3 = boto3.client("s3")
spark = GlueContext(SparkContext.getOrCreate()).spark_session
spark.conf.set("spark.sql.sources.partitionColumnTypeInference.enabled", "false")


def current_version() -> str:
    found = s3.list_objects_v2(Bucket=BUCKET, Prefix="v2/_manifests/", MaxKeys=1)
    return "v2" if found.get("KeyCount") else "v1"


def latest_ready(prefix: str) -> str:
    markers = [
        obj["Key"]
        for page in s3.get_paginator("list_objects_v2").paginate(Bucket=BUCKET, Prefix=prefix)
        for obj in page.get("Contents", [])
        if obj["Key"].endswith("/_SUCCESS")
    ]
    return max(markers).rsplit("/", 1)[0] + "/"


version = current_version()

conversations = spark.read.parquet(
    f"s3://{BUCKET}/" + latest_ready(f"{version}/datamart_claudia/claudia_conversation/")
)

messages = spark.read.parquet(f"s3://{BUCKET}/v1/datamart_claudia/claudia_messages/")
if version == "v2":
    messages = messages.unionByName(
        spark.read.parquet(f"s3://{BUCKET}/v2/datamart_claudia/claudia_messages/"),
        allowMissingColumns=True,
    )

delivered = (
    F.when(F.col("etl_date") == "__initial__", F.lit("0000-00-00")).otherwise(F.col("etl_date")),
    F.coalesce(F.col("etl_hour"), F.lit("")),
)
latest = Window.partitionBy("message_id").orderBy(*[c.desc() for c in delivered])
messages = (
    messages.withColumn("_rank", F.row_number().over(latest))
    .filter("_rank = 1")
    .drop("_rank")
)

• Tables rewritten in full, such as claudia_conversation: latest_ready returns the most recent folder holding _SUCCESS, which is a date in v1/ and a date and hour in v2/. The same call works on both sides of the switch.
• Tables delivered in parts, such as claudia_messages: the history in v1/ is always read, and the folders in v2/ are added once they exist. The example reads the whole table on every run for clarity. In production, read the v1/ history once and, from then on, only the v2/ folders you haven’t processed yet.
• Keep the latest version of each row. A row of claudia_messages or claudia_prompt_revision that changes after delivery arrives again in a later folder. The window at the end keeps, for each message_id, the row from the most recent folder. In an incremental pipeline, the same rule is an upsert by key.
• Partition columns are read as strings. etl_date holds the literal __initial__ in v1/, and turning type inference off keeps etl_date, etl_hour and snapshot_hour as strings on both sides, so the union lines up.
• Permissions don’t change. The access you already have covers the whole bucket, v2/ included.