Following are the details of the requirement:

1.      I am using databricks notebook to read data from Kafka topic and writing into ADLS Gen2 container i.e., my landing layer.

2.      I am using Spark code to read data from Kafka and write into landing layer.

3.      Next step is, I am reading Json files from landing layer and moving to bronze layer, which is another container in my ADLS Gen2. For this purpose, I am using Autoloader with Delta Live table to create table using Autoloader.

Here is the code for the same:

@dlt.table(

 name = tablename,

 comment = "Create Bronze Table",

 table_properties={

  "quality": "bronze"

 }

)

def Bronze_Table_Create():

 return (

  spark

  .readStream

  .format("cloudFiles")

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

  .option("cloudFiles.schemaLocation", schemalocation) 

  .option("cloudFiles.inferColumnTypes", "true")

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

  .load(sourcelocation)

 )

4.      This code works fine for me, it infers the schema as well. However, I have one scenario, which I am trying to handle. Which I mentioned step by step below:

i.                    I want to validate the schema, so that if there is any change in the schema, I should get notified and the job should get failed. I can handle it through SchemaEvolutionMode. However, my scenario is quite different. In my scenario, I am having one column RawData, which is of type object and there is not specified schema for it. It will get dynamic values, because of that if I infer the schema and apply schema validation, then every time it will bring new schema and throw schema mismatch error.

ii.                  Is there is any solution by which I can exclude the RawData column from schema validation, so that I allows this column to have any type of data.

I am struggling for this from a long time. Any help on this is helpful. Please let me know if any additional details are required on this.

Sample Json:

{

 "Header": {

   "SchemaVersion": "1.0",

   "EventId": "123",

   "EventTime_UTC": "2022-09-22 16:18:16",

   "Environment": "dev",

 },

 "Payload": {

   "RawData": {

     "CusID": "12345",

     "Status": "Pending",

     "LastModifiedAt": "2022-09-22 16:18:12",

     "ContainerName": "default,

     "CreatedAt": "2022-09-22 16:18:11" *The data in the RawData is inconsistent like it can have different columns*

   },

   "Data": {

     "CustID": "12345",

     "ArrayKeys": [

       {

         "ArrayName": "WorkHistory",

         "ArrayKeyName": "SampleId"

       }

     ]

   }

 }

}