@uzair mustafa​ : I am giving a framework to think about based on parallelizing the S3 file processing, Spark's caching capabilities and using Delta Lake's time travel capabilities, please see if this helps you to get started

from pyspark.sql.functions import col
from pyspark.sql.types import StructType, StructField, StringType
 
# Define the schema for the S3 files
schema = StructType([
  StructField("timestamp", StringType()),
  StructField("data", StringType())
])
 
# Read the list of S3 file paths from DynamoDB
file_paths = read_file_paths_from_dynamo_db()
 
# Filter out the file paths that have already been processed
delta_table = DeltaTable.forPath(spark, "delta_table_path")
processed_files = delta_table.toDF().select("file_path").distinct()
file_paths = [f for f in file_paths if f not in processed_files]
 
# Parallelize the file processing
file_paths_rdd = sc.parallelize(file_paths, len(file_paths))
data_rdd = file_paths_rdd.map(lambda f: (f, read_data_from_s3(f)))
data_df = data_rdd.toDF(["file_path", "data"]).select(
  col("file_path"),
  col("data.timestamp").alias("timestamp"),
  col("data.data").alias("data")
)
 
# Write the data to the Delta table
data_df.write.format("delta").mode("append").save("delta_table_path")

In this example, we read the list of S3 file paths from DynamoDB and filter out the files that have already been processed by querying a Delta table. Then, we parallelize the file processing using Spark's parallelize method and map function. Finally, we write the results to a Delta table using the write method. Note that the read_data_from_s3 function is assumed to be provided by the client, and should be modified to process the S3 data in parallel.