Kutbuddin
New Contributor III

‎05-27-2024 08:10 PM

Sharing code example:
1. Read stream from mongo
2. Write stream to delta table.
3. Insert into a internal checkpoint table.

# COMMAND 
query_cdc = (
spark.readStream.format("mongodb")
.option(
"spark.mongodb.connection.uri",
f"mongodb://{mongodb_user}:{mongodb_password}@{mongodb_host}/{source_database}.{source_collection}?authSource={mongodb_authsource}",
)
.option("spark.mongodb.database", source_database)
.option("spark.mongodb.collection", source_collection)
.option("spark.mongodb.change.stream.lookup.full.document", "updateLookup")
.option("spark.mongodb.read.aggregation.pipeline", "")
.schema(
{OurSchema}
)
.option("forceDeleteTempCheckpointLocation", "false")
.option("outputExtendedJson", "true")
.load()
)
# COMMAND
dt=int(datetime.datetime.strftime(datetime.datetime.now(),'%Y%m%d%H%M%S'))

query=(query_cdc.withColumn("ingestionTime", lit(dt)).writeStream \
.trigger(availableNow=True)
.partitionBy("ingestionTime")
.option("mergeSchema", "true")
.outputMode("append")
.option("checkpointLocation", target_bronze_checkpoint)
.toTable(f"{target_bronze_database}.{target_bronze_table}"));
 
 
# COMMAND
spark.sql(f"insert into table {internal_tab_name} values({dt})")
0 Kudos