Loading Multiple CSV files across all subfolder levels with Wildcard file name

Question

I want to Load Multiple CSV files matching certain names into a dataframe. Currently i am looping through the whole folder and creating a list of filenames and then loading those csv's into the dataframe list and then concatenating that dataframe.

The approach i want to use (if possible) is to bypass all the code and read all files in a one liner kind of approach.

I know this can be done easily for single level of subfolders, but my subfolder structure is as follows

Root Folder
    |
    Subfolder1
          |
          Subfolder 2
                |
                 X01.csv
                 Y01.csv
                 Z01.csv
    |
    Subfolder3
          |
          Subfolder4
                |
                X01.csv
                Y01.csv
    |
    Subfolder5
          |
          X01.csv
          Y01.csv

I want to read all "X01.csv" files while reading from Root Folder. Is there a way i can read all the required files in code something like the below

filepath = "rootpath" + "/**/X*.csv"   
df = spark.read.format("com.databricks.spark.csv").option("recursiveFilelookup","true").option("header","true").load(filepath)

This code works fine for single level of subfolders, is there any equivalent of this for multi level folders ? i thought the "recursiveFilelookup" option would look across all levels of subfolders, but apparently this is not the way it works.

Currently i am getting a

Path not found ... filepath

exception

any help please

Ingwersen_erik · Answer

Have you tried using the glob.glob function?

You can use it to search for files that match certain criteria inside a root path, and pass the list of files it finds to spark.read.csv function.

For example, I've recreated the folder structure from your example inside a Google Colab environment:

To get a list of all CSV files matching the criteria you've specified, you can use the following code:


import glob


rootpath = './Root Folder/'
# The following line of code looks through all files
# inside the rootpath recursively, trying to match the
# pattern specified. In this case, it tries to find any
# CSV file that starts with the letters X, Y, or Z,
# and ends with 2 numbers (ranging from 0 to 9).
glob.glob(rootpath + "**/[X|Y|Z][0-9][0-9].csv", recursive=True)

# Returns:
# ['./Root Folder/Subfolder5/Y01.csv',
#  './Root Folder/Subfolder5/X01.csv',
#  './Root Folder/Subfolder1/Subfolder 2/Y01.csv',
#  './Root Folder/Subfolder1/Subfolder 2/Z01.csv',
#  './Root Folder/Subfolder1/Subfolder 2/X01.csv']

Now you can combine this with spark.read.csv capability of reading a list of files to get the answer you're looking for:


import glob
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

rootpath = './Root Folder/'
spark.read.csv(glob.glob(rootpath + "**/[X|Y|Z][0-9][0-9].csv", recursive=True), inferSchema=True, header=True)

Note

You can specify more general patterns like:

glob.glob(rootpath + "**/*.csv", recursive=True)

To return a list of all csv files inside any subdirectory of rootpath.

Additionally, to consider only the immediate subdirectories files, you could use something like:

glob.glob(rootpath + "*.csv", recursive=True)

Edit

Based on your comments to this answer, does something like this works on Databricks?


from notebookutils import mssparkutils as ms
# databricks has a module called dbutils.fs.ls
# that works similarly to mssparkutils.fs, based on
# the following page of its documentation:
# https://docs.databricks.com/dev-tools/databricks-utils.html#ls-command-dbutilsfsls

def scan_dir(
    initial_path: str,
    search_str: str,
    account_name: str,
):
    """Scan a directory and subdirectories for a string.

    Parameters
    ----------
    initial_path : str
        The path to start the search. Accepts either a valid container name,
        or the entire connection string.
    search_str : str
        The string to search.
    account_name : str
        The name of the account to access the container folders.
        This value is only used, when the `initial_path`, doesn't
        conform with the format: "abfss://@.dfs.core.windows.net/"

    Raises
    ------
    FileNotFoundError
        If the `initial_path` informed doesn't exist.
    ValueError
        If `initial_path` is not a string.
    """
    if not isinstance(initial_path, str):
        raise ValueError(
            f'`initial_path` needs to be of type string, not {type(initial_path)}'
        )
    elif not initial_path.startswith('abfss'):
        initial_path = f'abfss://{initial_path}@{account_name}.dfs.core.windows.net/'
    try:
        fdirs = ms.fs.ls(initial_path)
    except Py4JJavaError as exc:
        raise FileNotFoundError(
            f'The path you informed \"{initial_path}\" doesn\'t exist'
        ) from exc
    found = []
    for path in fdirs:
        p = path.path
        if path.isDir:
            found = [*found, *scan_dir(p, search_str)]
        if search_str.lower() in path.name.lower():
            # print(p.split('.net')[-1])
            found = [*found, p.replace(path.name, "")]
    return list(set(found))

Example:

# Change .parquet to .csv
spark.read.parquet(*scan_dir("abfss://CONTAINER_NAME@ACCOUNTNAME.dfs.core.windows.net/ROOT/FOLDER/", ".parquet"))

This method above worked for on Azure Synapse:

Loading Multiple CSV files across all subfolder levels with Wildcard file name

1 Answers

Note

Edit