[dagster-airlift] create a custom operator for lakehouse example

[dpeng817]

As discussed offline, adds a custom operator for the lakehouse example. This should allow us to better demonstrate the step of "writing an operator to translate dagster code."

I think this leaves us with an unclear interstitial step for this type of asset. What's the "observe" phase look like here? Do we expect people to just provide an assetspec? We should sketch that out in a future PR.

[dpeng817]

• [dagster-airlift] annotated dbt example #23634
• [dagster-airlift] peer/observe/migrate for dbt example #23614
• [dagster-airlift] more flexible DbtProjectDefs params #23618
• [dagster-airlift] create a custom operator for lakehouse example #23597 👈
• [dagster-airlift] change dbt example to include migration fxnality #23592
• [dagster-airlift] rename peering to dbt example #23591
• [dagster-airlift] Fix sensor for migrated tasks #23587
• [dagster-airlift] fix munging of unserializable metadata #23586
• master

This stack of pull requests is managed by Graphite. Learn more about stacking.

Join @dpeng817 and the rest of your teammates on Graphite

[benpankow]

left a clarification q, but code seems good to me

[benpankow]

The main purpose of the custom DefsFactory here is to produce an API/user code matching the custom Airflow operator, is that right? e.g. there's no technical reason that there needs to be a 1:1 mapping of operator to defs factory - just making sure I understand this right.

It seems like a user could just use a PythonDefs here with a fn/lambda to define the inputs to load_csv_to_duckdb, but the downside is that there's a higher mental load in migrating that way since the user code ends up looking less similar moving airflow->Dagster?

load_iris = PythonOperator(
  task_id="load_iris", 
  dag=dag,
  python_callable=lambda: load_csv_to_duckdb(
    table_name="iris_lakehouse_table",
    csv_path=Path("iris.csv"),
    duckdb_path=Path(os.environ["AIRFLOW_HOME"]) / "jaffle_shop.duckdb",
    column_names=[
        "sepal_length_cm",
        "sepal_width_cm",
        "petal_length_cm",
        "petal_width_cm",
        "species",
    ],
    duckdb_schema="iris_dataset",
    duckdb_database_name="jaffle_shop",
  )
)

[dpeng817]

You're correct that PythonDefs could be used here, in the same way that a PythonOperator could be used to easily achieve the same goal of the load_csv_etc operator we use. The goal is to make it explicitly clear this semi 1:1 mapping exists for your custom operators / provide a clear path forward. The example is contrived, but the pattern isn't

[benpankow]

Thanks, that makes sense 👍