Updating dazfuller suggestion, but including code for one level of partitioning,

of course if you have deeper partitions then you will have to make a function and do a recursive call to get to the final directory containing parquet files.

Parquet will have the schema for everything but the partitions, you will have to add that,

I chose the datatype for partitions to be strings, not sure you could deduce more than that, and honestly not sure it matters in terms of spark optimizations.

Here is the code..

# create a working directory
 
data_path = "/tmp/directory_example"
# DELETE CONTENT 
# useful if run more than once or altered
dbutils.fs.rm(data_path, True)
dbutils.fs.mkdirs(data_path)

# create a database
database_name = "tomh_directory_test"
 
# CLEAR DATABASE IF ALREADY EXISTS
spark.sql(f"drop database IF EXISTS {database_name} cascade ")
spark.sql(f"create database {database_name}")
spark.sql(f"use {database_name}")

Create a dataframe to write sample content.

I partition in two different ways.

You might want to test if you have deeper nested partitions for example.

schema = "ID int, NAME String, DEPARTMENT String, YEAR string"
data = [[1,"Charles", "training","2020"],[2,"Evan", "sales", "2021"],[3,"Dorothy", "training", "2020"],[4,"Jason", "sales", "2021"]]
df = spark.createDataFrame(data, schema).coalesce(1)
display(df)

Write the data partitioned by department

table_name = "/table1"
partition_path = f"{data_path}{table_name}"
print(partition_path)
df.write.format("parquet").partitionBy("DEPARTMENT").save(partition_path)

Write the data partitioned by year

table_name = "/table2"
partition_path = f"{data_path}{table_name}"
print(partition_path)
df.write.format("parquet").partitionBy("YEAR").save(partition_path)

Current status

We have a directory with two subdirectories, each partitioned by a different column

Next steps

Write some code to generate statements similar to this

create table table1 (ID int, NAME string, DEPARTMENT STRING, YEAR STRING) USING PARQUET partitioned by (DEPARTMENT) location "/tmp/directory_example/table1";

It would be up to you if you wanted to add an "if not exists clause" to the create table statements. Only new directories would have tables created.

Table Names

I use the directory for the table name

Partitions

If the parquet files have subdirectories, those will be our partitions, I am only going one level deep here, you may need to recurse till you hit the data files if some tables have multiple partitions

Table Location

This will be the directory same as the one for the table name

Table Schema

Table Schema will be a combination of a schema generated by reading one of the partition folders and adding the partition column.

note if empty partitions exist you will have to catch that and read another partition

Create table

We will use external tables, by defining the location the tables are external.

MSCK

The code runs MSCK repair table to update the metastore to check for partitions.

basedir = dbutils.fs.ls(data_path)
#print(basedir)
#partition_list = []
for x in basedir:
  if x[1].endswith("/"): # has a subdirectory
    # Use directory name for table name
    # basedir is a list of row objects
    # item at index 1 x[1] is file name
    table_name = x[1][:-1] # remove last character /
    print(f"table name: {table_name}")
    
    # Get table location from same list of file objects
    # x[0] is full path
    
    table_location = x[0]
    print(f"table location: {table_location}")
    
    # To get the partition column get list of row objects 
    # from dbutils.fs on the subdirectory
    # Same logic as above, get directory name and parse for name of partition column
    path_to_one_subdirectory = dbutils.fs.ls(x[0])[0][0] # this will be a directory with parquet files you will need a recursive call if sub-partitions are used
    partition_column = (dbutils.fs.ls(x[0])[0][1].split("=")[0])
    print(f"partition column is: {partition_column}")
    
    # Get the base schema
    # Create a dataframe from the data directories, the ones that hold parquet files
    df = spark.read.format("parquet").load(path_to_one_subdirectory)
    base_schema = df.schema.simpleString()[7:-1].replace(":"," ")
    #print(myschema[7:-1].replace(":"," "))
    #print(base_schema)
  
    # Add the partition column
    full_schema = f"{base_schema} , {partition_column} string"
    #print("######")
    print(f"full schema is: {full_schema}")
    #print("#######")
    
    # Build the create table statements
    create_table_statement = f"create table {table_name} ({full_schema}) USING PARQUET partitioned by ({partition_column})  location '{table_location}'"
    print(f"Create table statement: {create_table_statement}")
    
    # create the table
    spark.sql(create_table_statement)
    spark.sql(f"msck repair table {table_name}")
    print("----")

At the end of that, you should have table1 and table2 in the current database.

%sql
 
SELECT * from table2;

%sql
 
SELECT * from table1;

To cleanup this code if you ran it all as an experiment do this.

#cleanup
 
dbutils.fs.rm(data_path, True)

#cleanup
 
spark.sql("drop database tomh_directory_test cascade ")

I have attached a notebook of this demo.