Large Data ingestion issue using auto loader

bzh · ‎08-07-2023

The goal of this project is to ingest 1000+ files (100MB per file) from S3 into Databricks. Since this will be incremental changes, we are using Autoloader for continued ingestion and transformation using a cluster (i3.xlarge).

The current process is very slow. Feels like it might take days to complete.

1. Each file is about 100,000+ rows. But when I run the code, Spark UI shows there are over 3,000,000 rows (see attachment) being processed altogether even though I setup maxFilesPerTrigger=1

2. We are using UDFs. We understand UDFs can be a costly operation compared to the Pyspark dataframe. But some of the python logic is on row level transformation which is quite hard to convert into a dataframe.

Here is the flow of our code:

# read stream from s3 mount
df = (
spark.readStream.format("cloudFiles")
.option("cloudFiles.format", "csv")
.option("sep", "||||")
.schema(transaction_schema)
.option("maxFilesPerTrigger", 1)
.load("dbfs:/mnt/s3/public/test_transactions")
)

# load contracts_df once outside the streaming operation
contracts_df = spark.read.table("hive_metastore.spindl.contracts")

# apply transformations to the entire streaming DataFrame
df = df.withColumn("transaction_hash", F.col("id"))
# ... more transformations ...

# define the UDFs
contract_info_udf = F.udf(contract_info, ...

# apply the UDFs
df = df.withColumn("contract_info", contract_info_udf(F.struct(contracts_df.columns)))
# ... more transformations ...

# write into transactions table
df.write.mode("append").insertInto("hive_metastore.spindl.test_transactions")

# write the stream
query = df.writeStream\
.format("delta")\
.option("checkpointLocation", "/tmp/delta/test/_write_stream_checkpoints/")\
.start()

query.awaitTermination()

Here is the flow of our code.