What I'm trying to achieve: ingest files into bronze tables with Autoloader, then produce Kafka messages for each file ingested using a DLT sink.

The issue: latency between file ingested and message produced get exponentially higher the more tables are added to the pipeline and simultaneous files are sent to be ingested; even at a low number of tables and a single file for each, latency can get to over a minute.

---

The test pipeline is a very basic one, and it's being run in continuous mode:

import dlt
from pyspark.sql import functions as f

@dlt.table
def bronze_table():
    return (
        spark.readStream.format("cloudFiles")
        # Options
        .load("/path/to/volume")
        .withColumn("created_at", f.current_timestamp())
        .withColumn("file_name", f.col("_metadata.file_path"))
    )

dlt.create_sink("kafka_sink", "kafka", options=kafkaOptions)

@dlt.append_flow(name=f"kafka_flow", target="kafka_sink")
def kafka_sink():
    return dlt.read_stream("bronze_table").groupBy("file_name").agg(f.first("created_at").alias("created_at"))

The Autoloader source path only has a single, small file. To test out, I have a script that creates a new file at the source, start a Kafka consumer, then measure the time between the file created and a message received with matching data.

To get closer to the actual volume, I add tables and a respective append_flow to each, targeting the same sink. So there's one Kafka sink, X tables, and X append_flow invocations all pointing to the same sink. But after 10 tables or so, and creating a file for each to be consumed at once, the latency of the last message to arrive can get over a minute.

CPU and memory are comfortably low in usage. Even so, I've tested out several cluster configurations, not only in resources but in cluster types as well, and at some point I've even equaled number of tables and worker cores to absolutely rule out resource constraints. No discernible difference.

I've also ruled out issues with our Kafka service by testing producing & consuming messages through a separate Databricks notebook - on the same workspace, to also rule possible networking issues.

The only thing I haven't tried yet was to split ingestion and sink/producer into two different pipelines, but I understand that my test pipeline with 10 tables isn't particularly large.

Can someone help me find the possible bottleneck in this process?