How to loop through a folder of csv files in an asset

[JamarWhitfield]

Let's say you have two resources with csv files in different locations. How would you loop through the resources in an asset so that both of the csv files get ran one after another after materializing?

[garethbrickman]

Could you elaborate what you mean by the csv files getting "run" after materializing?

[JamarWhitfield]

The csv files would be stored in a directory named "csv_locations" and it would make in place changes to each csv file, to generate new csv files.

[garethbrickman]

Okay, so would that directory just be on your local system? How does that relate to your original description of "two resources with csv files in different locations"? Could you explain more what you meant by resources and locations?

(By the way, Dagster has a concept called a Resource so I want to be sure of your meaning)

[JamarWhitfield]

I may be confused sorry about that.

[garethbrickman]

No problem, Resource is a more advanced Dagster concept that we'll bypass using for now.

Here's an example where we just provide a config to the asset that will specify the path to the folder that has the csv files in it. The asset materialization will process all csv files in the folder and then create new files prefixed with "processed_" in their name.

import os
import pandas as pd
from dagster import asset

@asset(config_schema={"csv_directory": str})
def process_csv_files(context):
    csv_directory = context.op_config["csv_directory"]
    
    # List all CSV files in the directory
    csv_files = [f for f in os.listdir(csv_directory) if f.endswith('.csv')]
    
    for csv_file in csv_files:
        file_path = os.path.join(csv_directory, csv_file)
        
        # Read the CSV file
        df = pd.read_csv(file_path)
        
        # Process the CSV file (example processing logic)
        processed_df = df.copy()
        processed_df['new_column'] = df['existing_column'] * 2
        
        # Save the processed data to a new CSV file
        new_file_path = os.path.join(csv_directory, f"processed_{csv_file}")
        processed_df.to_csv(new_file_path, index=False)
        
        # Log the processed data
        context.log.info(f"Processed data saved to {new_file_path}")

To run the asset with the config that will contain the path to the csv files, you can specify the configuration directly in the Dagster UI.

1. Start Dagster by running:

   dagster dev

2. Open the Dagster UI at http://127.0.0.1:3000/asset-groups.

3. Find the process_csv_files asset and click on it.

4. Click on the "Launchpad" tab.

5. In the "Config" section, add the following configuration:

   ops:
     process_csv_files:
       config:
         csv_directory: "path/to/csv_locations"

6. Click the "Launch Run" button to materialize the asset.