Massive performance degradation for table writes one row at a time

[joshivinay]

We are using delta-rs to read large and deeply hierarchical json files (1 MB - 20 MB) and write them to a delta table in append only mode using write_delta().
Due to current system limitations, we are forced to have one container handle one file at a time. The number of containers running in parallel can go to about 1000.
Each container receives a pointer to a json file, it reads the json file and writes it the delta table. We have a backoff mechanism to backoff in case of conflicts. If there are 50000 json files, we end up with a delta table with 50,000 parquet files and subsequently 50,000 commit log files.
We have noticed that the performance degrades drastically despite checkpointing of the log files. We also see a humongous number of s3 getObject calls.
Some questions:

1. Since we are checkpointing , I would have assumed that for each write, we would read the transaction log checkpoint file, then the incremental json files. So a worst case write would be 1 checkpoint.parquet + 99 .json files per write.
2. When in append only mode, would any of the data parquet files even be loaded into the container? We are doing blind appends to the table using write_delta(). We dont load the table using DeltaTable.
3. After putting 30k files into the table which took hours and hours, writing even one file causes an OOM. It takes almost 30 minutes to write a single 2 mb file or it eventually hits an OOM.

I did a search for a similar discussion and found one where the number of object storage calls was high, but nothing like we are seeing. We raked up $$$ in a performance test just because of the s3 getObjects and the processing wasnt complete.
Glue eventually couldnt OPTIMIZE and VACUUM RETAIN 0 HOURS either after a 6 hour run.

We are trying to understand if there is any better way to do this or are these limitations with delta-rs for this kind of processing.

Pseudo code here.
`
import time
from dataclasses import dataclass, fields
from datetime import datetime
from typing import Any, Mapping

import pyarrow as pa
from deltalake import DeltaTable, write_deltalake
from deltalake.table import WriterProperties
from deltalake.writer import try_get_deltatable
from jsonref import replace_refs
from loguru import logger
from pyarrow import json as pa_json

# each write attempt is really 15 attempts, so this should be more than enough to handle high load
MAX_WRITE_ATTEMPTS = 15
BACK_OFF_DELAY_MILLISECONDS = 100

download the json file and the schema

data = download_json()
full_schema = replace_refs(data.schema, proxies=False)

logger.info("Converting to pyarrow schema")
data_schema: pa.Schema = get_pyarrow_schema(full_schema)

logger.info("Reading json file")
dataframe: pa.Table = pa_json.read_json(
    ids_data.download["download"],
    parse_options=pa_json.ParseOptions(
        explicit_schema=data_schema, unexpected_field_behavior="ignore"
    ),
    # set a really large block size so that pyarrow can read the whole file
    read_options=pa_json.ReadOptions(block_size=1024 * 1024 * 1024),
)

# add the extra columns to both the dataset and the schema
metadata = Metadata(
  # Some metadata
)

dataframe, data_schema = add_metadata_columns(dataframe, data_schema, metadata)

if not env.is_local:
    storage_options = {
        "AWS_S3_LOCKING_PROVIDER": "dynamodb",
        "DELTA_DYNAMO_TABLE_NAME": env.lakehouse_delta_log_table,
    }
    delta_table_path = f"s3a://{env.lakehouse_bucket}/{get_s3_path()}"
else:
    storage_options = {"AWS_S3_ALLOW_UNSAFE_RENAME": "true"}
    delta_table_path = (
        f"{env.local_prefix}/{env.lakehouse_bucket}/{get_s3_path()}"
    )

writer_props = WriterProperties(compression="SNAPPY")

if not try_get_deltatable(delta_table_path, None):
    configuration = {"delta.enableChangeDataFeed": "true"}
    logger.info("Table does not exist! Creating table at '{}'", delta_table_path)
    DeltaTable.create(
        table_uri=delta_table_path,
        schema=data_schema,
        configuration=configuration,
        storage_options=storage_options,
    )

# If many concurrent writes are happening, the write_deltalake function may reach its own max attempts (15)
# This loop will retry the write until it works.
try_num = 0
for try_num in range(MAX_WRITE_ATTEMPTS):
    logger.info("Trying to write to delta (try #{}): {}", try_num, delta_table_path)
    try:
        write_deltalake(
            delta_table_path,
            dataframe,
            schema=data_schema,
            mode="append",
            schema_mode="merge",
            storage_options=storage_options,
            writer_properties=writer_props,
            engine="rust",
        )
        logger.info("Done write!")
        return
    except:
        logger.exception("Try #{} failed to write", try_num)
        time.sleep(try_num * BACK_OFF_DELAY_MILLISECONDS)

logger.error("Exceeded max tries writing to delta")`

[joshivinay]

Forgot to mention. We are using delta-rs 0.18.0

[ion-elgreco]

Use v0.19

[ion-elgreco]

1,3. you are reading the checkpoint, but if you don't optimize compact properly, your checkpoint will be huge and you may get OOM if your container doesn't have enough resources.
2. When you write, a table state will always be loaded, you can't blindly write because we need to assess the schema, if you are allowed to write with your writer and so forth