Hi, i am trying to load mongo into s3 using pyspark 3.1.1 by reading them into a parquet.

My code snippets are like:

df = spark \

.read \

.format("mongo") \

.options(**read_options) \

.load(schema=schema)

df = df.coalesce(64)

write_df_to_delta(spark, df, s3_path)

read_count = df.count()

inserted_df = read_delta_to_df(spark, s3_path)

inserted_count = inserted_df.count()

all sparksession, mongo connection and s3 path configured well. What i found is that read_count and inserted_df count do not match, there is a gap of around 300-1200 rows. But my write to delta did not give me any error. I wonder why is this the case? what's causing it?

what i can see form rancher: 'read_count': 1373432, 'inserted_count': 1372492

def read_delta_to_df(

spark: SparkSession,

s3_path: str

) -> DataFrame:

log.info("Reading delta table from path {} to df".format(s3_path))

df = spark \

.read \

.format("delta") \

.load(s3_path)

return df

def write_df_to_delta(

spark: SparkSession,

df: DataFrame,

s3_path: str,

mode: Optional[str] = "overwrite",

partition_by: Optional[Union[str, List[str]]]= None,

retention: Optional[int] = 0

) -> None:

log.info("Writing df to delta table, {}".format(s3_path))

df.printSchema()

try:

df \

.write \

.format("delta") \

.mode(mode) \

.option("overwriteSchema", "true") \

.save(

path=s3_path,

partitionBy=partition_by)

except Exception as e:

log.error(f"error occured with error msg: {e}")