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}")
