Below function executes fine:

def autoload_to_table(data_source, source_format, table_name, checkpoint_directory):

  query = (spark.readStream

         .format("cloudFiles")

         .option("cloudFiles.format", source_format)

         .option("cloudFiles.schemaLocation", checkpoint_directory)

         .load(data_source)

         .writeStream

         .option("checkpointLocation", checkpoint_directory)

         .option("mergeSchema", "true")

         .table(table_name))

  return query

Receiving Error while calling function. Please let me know where is the problem.