OOM in pipeline running.

I just run pipeline with addId. But OOMKilled happens in the running. It happens all the time.
My worker has 80cpu 1.2TB memory. And I need to process 30k files which each file is below 512MB
Even I use multiple workers, there are OOMKilled all the time.

The code is below.

def process_dataset(input_path, output_path, dir_name):
    """
    为指定的数据集运行 AddId Pipeline
    """
    # Create pipeline
    pipeline = Pipeline(name=f"heuristic_filtering")
    
    # ============================= Add Id =============================
    reader = JsonlReader(
        file_paths=input_path,  # 直接使用目录路径，会自动递归查找所有 .jsonl 文件
        fields = ['text'],
        # blocksize="256MB",
        files_per_partition=8,
    )
    pipeline.add_stage(reader)
    pipeline.add_stage(AddId(
        id_field="doc_id",        # Field name for IDs
        id_prefix=f"{dir_name}_",         # Optional prefix
        overwrite=True,                 # Overwrite existing IDs
    ))

    # ============================= Writer =============================
    pipeline.add_stage(JsonlWriter(path=output_path))

    # Run with Xenna (recommended)
    executor = XennaExecutor(
        config={
            "execution_mode": "streaming",
            "ignore_failures": True,
            "reset_workers_on_failure": True,
            "cpu_allocation_percentage": 0.5,
            "max_workers_per_stage": 8,
            "autoscale_interval_s": 10 ** 10,
            "logging_interval": 300
        }
    )
    results = pipeline.run(executor)

Start with the provided process_dataset example and inspect how JsonlReader, AddId, JsonlWriter, and XennaExecutor interact under streaming execution. Reproduce the run with the stated 30k-file workload and executor settings, then trace worker memory usage and failure behavior. Done means identifying the cause of the repeated OOMKilled failures and documenting or validating a stable configuration or fix.