Databricks jobs for assets?

[geoHeil]

https://github.com/dagster-io/hooli-data-eng-pipelines/blob/3b43ceb2ae4bb20acc014ab2d305e238294b11a4/hooli_data_eng/assets/forecasting/__init__.py#L183 is using submitted jobs (looks like databricks jobs are used)

https://github.com/dagster-io/hooli-data-eng-pipelines/blob/3b43ceb2ae4bb20acc014ab2d305e238294b11a4/hooli_data_eng/resources/databricks.py a step launcher (looks like an interactive cluster is created)

How can I ensure that the cheaper DBU pricing (jobs instead of interactive) is being used when dagster is interacting with databricks?

[zyd14]

Both of these examples use job clusters, the lower DBU-priced resource.

In the first example, the submit task call here defines a new_cluster key, which corresponds to the new_cluster key in the Databricks jobs REST API for creating a job cluster to use to carry out the submitted job.

In the latter example, where a step launcher is being used, the cluster config key in the step launcher run_config has two values: new for defining new job clusters (the lower DBU pricing), and existing for pointing to an interactive cluster (higher DBU pricing). It is not possible to create a new interactive cluster and then run the job on it through the step launcher, so if you're not pointing your config to an existing cluster, then a job cluster is being created to carry out the submitted job.

So in the case of https://github.com/dagster-io/hooli-data-eng-pipelines/blob/3b43ceb2ae4bb20acc014ab2d305e238294b11a4/hooli_data_eng/resources/databricks.py where the step launcher is being used, the config on line 33 specifies that a new job cluster should be created to carry out the job.

[weberdavid]

@zyd14 We are using the second option - the step launcher. Is it possible to point the config to an existing job cluster? We want to use the same job cluster for multiple jobs and not a new cluster for every job / run.

At the moment we use an existing interactive cluster, but would love to go with the lower DBU pricing.

[zyd14]

Unfortunately that is not currently possible. Databricks only allows one to reuse a job cluster for tasks that are in the same Databricks workflow. A Dagster step which is configured with the databricks_pyspark_step_launcher maps to a single Databricks workflow with a single task. Put differently, Databricks does not allow one to keep a job cluster alive between workflows, and every Dagster step configured with the databricks_pyspark_step_launcher is submitted to Databricks in its own workflow.

I too would like to be able to reuse job clusters across steps in a Dagster job, but it seems that the step launcher may not be the right abstraction. It's a bit difficult to imagine how one could translate a Dagster job to a Databricks workflow with multiple steps and cluster reuse enabled considering the possibility that some of the steps of the Dagster job may not need to be executed on Databricks or need to execute on another platform entirely (such as snowflake).

One path forward could be a DatabricksRunLauncher implementation, but there are some caveats there in that it would likely need to assume that all steps in a Dagster run would need to execute on Databricks together as a workflow. It would also probably require a way to opt out of running a Dagster job on Databricks, as RunLaunchers are configured per-deployment, and I find it unlikely that most people would want to run all their Dagster jobs on Databricks (maybe that's a biased assumption by me though as I'm always trying to find ways to run less stuff on Databricks).

[geoHeil]

Agreed!

[geoHeil]

When will the cluster be started/stopped? If there are multiple assets and each needs their own databricks cluster can I use a different cluster for each asset? If the cluster should be reused for (smaller) assets is it possible to keep it on/running?

[zyd14]

A cluster is started and stopped for each step configured with the databricks_pyspark_step_launcher. To configured different clusters with different resources for different assets you would have a different step launcher instance configured on each of the different steps:

@asset(required_resource_keys={"step_launcher_A"})
def asset_with_cluster_A():
  ...

@asset(required_resource_keys={"step_launcher_B"})
def asset_with_cluster_B():
  ...

asset_job = with_resources([asset_with_cluster_A, asset_with_cluster_B], 
                            resource_defs={"step_launcher_A": databricks_pyspark_step_launcher.configured(...), "step_launcher_B": databricks_pyspark_step_launcher.configured(...)})

It's currently not possible to reuse a job cluster across multiple steps (see my answer to @weberdavid above). If speed is the requirement here, you'd have to run the steps on an existing interactive cluster, but then that costs more DBUs. However, you could configure some smaller steps to happen on a shared interactive cluster and then have larger steps launch their own job cluster.

[geoHeil]

What exactly constitutes a step? I.e. is one asset one step?

[zyd14]

Yes in this context each asset or op configured with the databricks_pyspark_step_launcher is an individual step.

[geoHeil]

@zyd14

for me the step launcher is showing up inside the workflows tab (which would be a non-interactive DBU cluster) is this the same case for oyu?

[zyd14]

Yes, using the

run_config:
  cluster:
    new:
      ...

step launcher config will launch a job cluster and not an interactive cluster. To launch a Dagster job on an existing interactive cluster, you must use the following config:

run_config:
  cluster:
    existing: <existing_cluster_ID>

[geoHeil]

But this is great ( or in fact exactly what I would want to have) this means this would keep the cluster alive using the step launcher - and give cheaper DBU.

Is this right?

If yes, for how long would the step launcher keep its cluster alive if asset A (used the steplauncher) B ( is another non steplauncher task with i..e a pipes workflow submission) and C (again a steplauncher)

[geoHeil]

Wait - but isn`t that conflicting with #17805 (reply in thread) your point above? It looks like it will run in both modes (depending onf if new or existing is chosen).

[zyd14]

So a little bit of nomenclature might be helpful to introduce here to improve communication in the future. The higher DBU cluster option we can refer to as interactive clusters, and the lower DBU cluster option we can refer to as job clusters, in alignment with Databricks' terminology.

To clarify, the existing config option must point to an existing interactive cluster. The new config option will create a job cluster for the step marked with the step launcher. The step launcher does not do anything to keep a job cluster alive, and multiple steps in a graph marked with step launchers will always launch new job clusters for each step.

To reiterate my previous comments there is currently no way to re-use job clusters for multiple steps. When using the step launcher you must choose between having each step spin up a new job cluster (lower DBU, longer initialization time) or using an existing interactive cluster (higher DBU, shorter step initialization time).

My team at Empirico has been exploring solutions to this, but the solution we're working on involves deploying another service external to Dagster and Databricks entirely to manage and monitor job clusters and is quite complex, as you could imagine there are many desirable ways to customize such functionality and many potential failure states. The step launcher itself doesn't have the scope and context necessary to manage a job cluster across multiple steps.

If a tool was built entirely within Dagster to manage multiple steps executing on a single cluster, it's likely that the appropriate abstraction to create would either be a custom executor or run launcher. It seems entirely possible that a run launcher could be doable, where the Dagster graph could be translated in to a Databricks workflow with the cluster reuse option enabled. The issue with that is that run launchers are scoped to a deployment, so unless the user was okay with running all jobs in a deployment on Databricks there would need to be a mechanism for indicating when a run should be launched on Databricks, and when it should fall back to a different run launcher. It also would only be able to launch entire runs on Databricks and would require step launchers to run steps outside of Databricks, which may not be desirable. The executor abstraction seems like it might be a way to have more fine-grained control over launching groups of steps on Databricks, but I'm not as familiar with it. I hope to set aside some time over the holiday break to explore both of these options a little more fully.

[geoHeil]

Thanks for this clarification!

What is then in your opinion an upside in choosing the step launcher vs. the pipes option? (except for using the existing cluster)?