We are building a DLT pipeline and the autoloader is handling schema evolution fine. However, further down the pipeline we are trying to load that streamed data with the apply_changes() function into a new table and, from the looks of it, doesn't seem to handle row updates with a new schema. However, during "Setting Up Tables" it fails with an "org.apache.spark.sql.catalyst.parser.ParseException" error. The only explanation I can think of is it doesn't like replacing a column field of type "Null" with "Struct".
Here is the code:
@dlt.view(
name = "authenticators_stream"
)
@dlt.expect_all_or_drop({"valid_doc": "doc IS NOT NULL"})
def stream_table():
return (
spark.readStream \
.format("cloudFiles") \
.option("cloudFiles.useNotifications", "true") \
.option("cloudFiles.queueUrl", "https://sqs.us-east-1.amazonaws.com/********/mongo-data-queue-testing") \
.option("cloudFiles.includeExistingFiles", "true") \
.option("cloudFiles.format", "json") \
.option("cloudFiles.inferColumnTypes", "true")
.option("cloudFiles.schemaEvolutionMode", "addNewColumns") \
.option("multiline","false") \
.option("cloudFiles.schemaHints", "_id STRING, ot STRING, ts TIMESTAMP, year INT, month INT, day INT")
.load(json_path))
dlt.create_streaming_live_table(
name = "authenticators_raw",
spark_conf = {"spark.databricks.delta.schema.autoMerge.enabled": "true"}
)
dlt.apply_changes(
target = "authenticators_raw",
source = "authenticators_stream",
keys = ["_id"],
sequence_by = F.col("ts"),
stored_as_scd_type = 2
)
And here is the full error message:
org.apache.spark.sql.catalyst.parser.ParseException:
[PARSE_SYNTAX_ERROR] Syntax error at or near '<'(line 1, pos 6)
== SQL ==
struct<__v:bigint,_id:string,buttonlabel:string,company:string,configuration:struct<parameters:struct<company-id:string,cyberarkurl:string,duo-sso-url:string,email:string,google-oauth-url:string,login-success-text:string,login-url:string,microsofturl:string,okta-url:string,oktasubdomain:string,onelogin-url:string,password:string,payroll-cookies-wait-for-url:string,payroll-provider-selector:string,ping-identity-url:string,request-id:string,secureid-url:string,subdomain:string,target-computing-resources-url:string,username:string,usersname:string,wait-for-milliseconds-param-key:string,wait-for-xpath-after-navigate:string,workday-organization-group-name:string>>,connector:string,createdat:string,optional:boolean,updatedat:string>
------^^^
