Hi, We want to understand what is the best practice for running solids with configurations files. Our simple use case is that we have a pipeline that we need to run on different files (each file with a different configuration). From this https://dagster.phacility.com/D2060 link and this PR #2049 we are not sure what is the preferred direction to take (the former one is more appealing to our use case). Any guidance would be appreciated.

So these are sort of different use cases. Could you elaborate more precisely on what you mean by "running on different files with different configs"?

A guess it's more like the former one. Currently have a several python files that run with YAML configuration files for ETL processes. The configurations include how to run functions from installed packages (e.g. changing the args in dask.to_parquet) or custom funcitons that we have written that mainly manipulate dataframes (e.g. adds or removes fields in dataframes).

I would probably use this: https://docs.dagster.io/latest/tutorial/config this way you can set a default

dask.to_parquet

but always have the convenience to overwrite the default

config

in your yaml with :

solids:
  your_ETL_pipeline_fuction
    config:
      output: "dask.to_delta"

That’s right. The config system is what you are looking for and if you are looking for ways to hydrate/dehydrate these files via solid config you should check out the input hydration/output materialization configs in the tutorial as well.

Hi, We decided to work with

spark

since it seems to be a bit more mature in this project. Since we want to have a separate pipeline for each data source - and want to make the code generic as possible (and not write a

solid

for every API call in

pyspark

we wrote the following function that is called from the YAML pipeline, however it fails if the function is trying to use a new column that is generated from within the pipeline. 1. Is there a way to write such function that will work without using

collect

(the

df.cache()

did not help) 2. if not - is our only option is to write every step of the ETL as a separate

solid

for every API call (since we have many data sources with many transformations - it will not be optimal in maintaining and generating new pipelines.

@solid(

config={"base_func": Permissive(),

"transform": Permissive()})

def run_transform(context, df: DataFrame):

d  = {"base_func": pyspark.sql.dataframe.DataFrame, "transform": transform.py}

for name, module in d.items():

for func, attr in context.solid_config[name].items():

df = getattr(module, func)(df, **attr)

df  = df.cache()   # did not help

The transform.py is our implementation of

pyspark.sql.functions

The YAML pipeline file is

resources:

spark:

config:

spark_conf:

.....

solids:

read_file:

config:

path: "<hdfs://path/to/file.csv>"

format: "csv"

run_transform:

config:

base_func:

fillna: {value: {"date_col", "2199-12-31 00:00:00}}

dropna: {how: "any", subset: ['col1', 'col2']}

transform:

rename_cols: {renamed_cols: {col1: "colA"}}

assignF: {col: "new_col", func: "concat", attr: ["date_col", "colA"]}