topic Re: Autoloader includeExistingFiles with retry didn't update the schema in Data Engineering

Autoloader includeExistingFiles with retry didn't update the schema

ajithgaade — Tue, 02 Jul 2024 03:38:41 GMT

Hi,
written in pyspark.

databricks autoloader job with retry didn't merge/update the schema.

spark.readStream.format("cloudFiles")

.option("cloudFiles.format", "parquet")

.option("cloudFiles.schemaLocation", checkpoint_path)

.option("cloudFiles.includeExistingFiles", "false"

.load(source_path)

.writeStream

.queryName("write stream query")

.option("checkpointLocation", checkpoint_path)

.trigger(availableNow=True)

.forEachBatch(batch_operation)

.option("mergeSchema", True)

.start()

.awaitTermination()

Error:

Error while reading file s3://path
Caused by: UnknownFieldException: [UNKNOWN_FILED_EXCEPTION.NEW_FILEDS_IN_FILE] Encountered unknown fields during parsing: filed type, which can be fixed by an automatic retry: false

tried running couple of times. set retry = 2 for job and task as well.

please can you help?

Re: Autoloader includeExistingFiles with retry didn't update the schema

Witold — Tue, 02 Jul 2024 06:17:40 GMT

What happens if you enable rescue mode?

.option('cloudFiles.schemaEvolutionMode', 'rescue')

Re: Autoloader includeExistingFiles with retry didn't update the schema

Giri-Patcham — Tue, 02 Jul 2024 06:22:04 GMT

Hi @ajithgaade ,

If you are using a merge statement inside the forEachBatch function batch_operation then you have to use DBR 15.2 and above to evolve the schema
https://docs.databricks.com/en/delta/update-schema.html#automatic-schema-evolution-for-delta-lake-merge

Re: Autoloader includeExistingFiles with retry didn't update the schema

ajithgaade — Tue, 02 Jul 2024 14:12:09 GMT

@Witold

rescue option won't evolve the schema.
https://docs.databricks.com/en/ingestion/auto-loader/schema.html#:~:text=evolve%20data%20types.-,rescue,-Schema%20is%20never

my requirement is schema should evolve automatically

Re: Autoloader includeExistingFiles with retry didn't update the schema

ajithgaade — Tue, 02 Jul 2024 13:53:29 GMT

Hi @Giri-Patcham
batch operation doesn't has merge statement. I am dropping tables and recreating. Tried clearing checkpoint location many times and different options. Tried with DBR 15.3, No Luck.

Re: Autoloader includeExistingFiles with retry didn't update the schema

Giri-Patcham — Tue, 02 Jul 2024 15:10:42 GMT

@ajithgaade can you share the sample code inside the batch function ?

Re: Autoloader includeExistingFiles with retry didn't update the schema

ajithgaade — Tue, 02 Jul 2024 16:17:31 GMT

Below is the sample one

def batch_operation(df, batch_id😞

logger.info(f"batch_id: {batch_id}")

src_cnt = 0

filtr_cnt = 0

df = src_table_df \

.withColumn('load_ctl_key', lit(aws_lck)

src_columns = df.columns

# checksum text column is generated and added to the DataFrame.

if condition

df2 = create_chk_sum_txt(df=df, cols=col_info, exc_cols=exclude_columns)

else:

df2 = df

df3 = df2.filter(

col("id").isNull() | (length(col("id")) > 50) | (length(col("cd")) > 10) | \

(col("5yr").isNotNull() & col("5yr").cast("int").isNull())

)

# columns=df3.columns

logger.info("df3 completed")

df4 = df3.withColumn("error_reason", when(col("id").isNull(), "id is Null") \

.when(length(col("id")) > 50, "id exceeds column size in base table") \

.otherwise("Miscellaneous Error"))

# Filter valid records to be loaded to Base Table

df5 = df2.exceptAll(df3).withColumn("error_reason", lit(""))

ins_cnt = df5.count()

filtr_cnt = src_cnt - ins_cnt

logger.info('src_cnt: {}'.format(src_cnt))

logger.info('ins_cnt: {}'.format(ins_cnt))

logger.info('filtr_cnt: {}'.format(filtr_cnt))

select_df = df5.select(*col_info)

logger.info("select_df completed")

err_df = df4.select(*col_info)

logger.info("err_df completed")

logger.info("Final Schema before writing to UC:")

select_df.printSchema()

err_df.printSchema()

(

transformed_df.write

.mode("append")

.saveAsTable(f"catalog.schema.{trgt_tbl}")

)

(

err_df.write

.mode("append")

.saveAsTable(f"catalog.schema.{err_tbl}")

)

if int(load_ctl_key) != -1:

print("Pushing metrics to UC")

payload = {

"load_ctl_key": load_ctl_key,

"job_id": job_id,

"batch_id": batch_id,

"table": trgt_tbl,

"src_cnt": src_cnt,

"filter_cnt": filtr_cnt,

"ins_cnt": ins_cnt

}

print("Payload: {}".format(payload))

payload_df = spark.read.json(sc.parallelize([payload]))

display(payload_df)

payload_df.write.mode("append").saveAsTable(f"catalog.schema.audit")

Re: Autoloader includeExistingFiles with retry didn't update the schema

Giri-Patcham — Tue, 02 Jul 2024 17:18:01 GMT

@ajithgaade

can you try setting this conf

spark.conf.set("spark.databricks.delta.schema.autoMerge.enabled", "true")

Re: Autoloader includeExistingFiles with retry didn't update the schema

mtajmouati — Tue, 02 Jul 2024 20:26:16 GMT

Hello,

Try this :

from pyspark.sql import SparkSession # Initialize Spark session spark = SparkSession.builder \ .appName("Auto Loader Schema Evolution") \ .getOrCreate() # Source and checkpoint paths source_path = "s3://path" checkpoint_path = "/path/to/checkpoint" # Define batch processing function def batch_operation(df, epoch_id): # Perform your batch operations here # For example, write to Delta table with schema merge df.write \ .format("delta") \ .mode("append") \ .option("mergeSchema", "true") \ .save("/path/to/delta/table") # Read stream with schema evolution df = spark.readStream.format("cloudFiles") \ .option("cloudFiles.format", "parquet") \ .option("cloudFiles.schemaLocation", checkpoint_path) \ .option("cloudFiles.includeExistingFiles", "false") \ .option("cloudFiles.inferColumnTypes", "true") \ .load(source_path) # Write stream with schema merge query = df.writeStream \ .format("delta") \ .option("checkpointLocation", checkpoint_path) \ .trigger(availableNow=True) \ .foreachBatch(batch_operation) \ .option("mergeSchema", "true") \ .start() query.awaitTermination()

and try Setting Retry Policies

{ "tasks": [ { "task_key": "example-task", "notebook_task": { "notebook_path": "/path/to/your/notebook" }, "max_retries": 2, "min_retry_interval_millis": 60000, "retry_on_timeout": true } ] }