checkpoint changes not working on my databricks job
Options
- Mark as New
- Bookmark
- Subscribe
- Mute
- Subscribe to RSS Feed
- Permalink
- Report Inappropriate Content
01-11-2026 11:45 AM
Hi,
I do have a job processing kafka stream using kafka.readstream process and due to some issue we changed the checkpoint path to other path and it pulled all the records and later when i changed to the original checkpoint location it is not pulling the new records coming through kafka. Any clue what can be done to make it running again the way it was. Here is the code snippet.
curr_env = getenv()
config_path = "../configs/CDLZ_MEMBERS_Custom_Tables.yaml"
process_name = "CDLZ_MEMBERS_CUSTOM_Tables"
config = load_yaml_config(config_path)
source_kafka_topic = config[process_name]["KAFKA_CONFIG"]["input_kafka_topic"]
raw_table_name = config[process_name]["DELTA_TABLE_CONFIG"]["raw_delta_table"]
tables_to_read = config[process_name]["KAFKA_CONFIG"]["tables_to_read"]
raw_table_name = raw_table_name.replace("{env}", curr_env)
checkpoint = get_checkpoint_path(source_kafka_topic + "/" + process_name)
kafka_stream = (
spark.readStream.format("kafka")
.option("kafka.bootstrap.servers", kafka_bootstrap_servers)
.option("kafka.security.protocol", "SASL_SSL")
.option(
"kafka.sasl.jaas.config",
"kafkashaded.org.apache.kafka.common.security.plain.PlainLoginModule required username='{}' password='{}';".format(
username, password
),
)
.option("kafka.ssl.endpoint.identification.algorithm", "https")
.option("kafka.sasl.mechanism", "PLAIN")
.option("failOnDataLoss", "false")
.option("subscribe", source_kafka_topic)
.option("startingOffsets", "earliest") # read from specific timestamp (in ms)
.load()
)
query = (
df_stream.writeStream.format("delta")
.outputMode("append")
.trigger(availableNow=True)
.option("checkpointLocation", checkpoint)
.option("mergeSchema", "true")
.option("schemaCheck", "false")
.toTable(raw_table_name)
)
query.awaitTermination()
