How to use Spark cluster mode in dbt?

[uplsh580]

I want to use spark cluster mode in dbt.

When I use spark-submit on cli, just add --deploy-mode cluster option in command.
But I don't know how set cluster mode in dbt-spark.

Please somebody let me know that.

Thanks.

[JCZuurmond]

Hi @uplsh580 , I expect you to start a cluster other than using spark-submit.

A difference between dbt-spark and spark-submit is that dbt-spark communicates with a cluster between jobs/tasks. A spark-submit sends the jobs/task once and waits for it to finish.

Does this help?

[uplsh580]

Thank you for your answer.

[akashrn5]

Hi @JCZuurmond

I actually didn't get your reply on spark cluster mode clearly. Actually i wanted to discuss a scenario here where without DBT, spark cluster mode can basically help me address that issue.

### Scenario:
Without DBT:
Let's assume that there are some SQLs which could run for longer time like 5hrs. In this case if i submit the query in client mode, there could be chances of client machine going down or any other incidents which could lead to a failure of spark job. Let's say overall job time is 5hours, its been 4hrs since my job is running and the client machine goes down. This will lead to wastage of time and resources for 4hrs. So Solution would be to go with spark cluster mode which will remove the client dependency in this case.

With DBT:
With DBT, as of now there are connection types like http, thrift and session. Lets consider the above scenario, and lets say there are few models of DBT whose SQLs would run for longer time. In that case, if my client machine goes down where dbt run is triggered, it will lead to above issue. So cluster mode kind of solution helps to avoid these issues. Since dbt-spark doesn't support spark cluster mode, as it needs to maintain the state of the models to update in dbt artifacts, how can we handle this scenario?

Please give your inputs or views on this. I would like to get @Fleid , @dbeatty10 views on this.

[shashwatsai]

Thank you Akash for bringing this up.

Hi Doug,
We are also trying to brainstorm on something similar, currently as per our understanding the execution state is tightly coupled with the pod/machine where the dbt run was triggered, if the machine goes down, the state also gets lost.

[Discussion][Client Mode] Problems we are trying to get better understanding.

1. State gets checkpointed after every model completion to a centralised object store or something similar.
2. The run gets resumed from the model where DAG failed, by provided the path of the state checkpointed in [1]

In the aforementioned example, a and b should not get recomputed, if the run_results are lost we will have to restart the DAG from starting.

As you have already pointed out the retry mechanism in v1.6, does it utilise run_results.json in the client machine where dbt run was triggered, do we see any benefit of decoupling the serialised state from client machine?

[Discussion][Cluster Mode | External Trigger]

We use a schedular

1. in step 1. at time t1, we spawn a machine where dbt run will happen, fetch dbt project files from artifacts store, this returns an identifier to external schedular.
2. client machine interacts with cluster through thrift or something similar.
3. write the model completion state to state store, checkpoint the state. Network partitioning happened and pod failed.
4. spawn new client machine at time t4, get artifacts, previous state from state store from dbt run id returned in step[1].
5. client machine interacts with cluster and starts run from failed model.
6. write the model completion state to state store.

please pardon, if this does not fall into the spark adapter scope, we can start a new issue in dbt core, if concerns look valid to you?

[uplsh580]

Because of client-side accidents that @akashrn5 said, or client-side resource limit, I have to use Spark Cluster mode.

I experimented with using DBT to meet the requirements of using Spark Cluster mode.
The simple architecture is as follows. (Use DBT, Airflow, and Spark on Yarn. (w/o k8s))

1. Parse the manifest.json file generated as a result of dbt-compile and set the dependency with the Airflow DAG.
   • ref. https://www.astronomer.io/blog/airflow-dbt-1/
2. Run DBT-Compile every day to create a date-updated SQL File.
3. Run Spark Cluster mode by including compiled SQL file in PySpark scripts and running Spark Submit.

In this case, only dbt-compile was used, so there was actually a limit to using many of the features of DBT.
Because this architecture has several limit, I decided not to apply it as it is to production operation.
I'm wondering if there's a better way.

@akashrn5 Thank you so much to start this discussion.

[akashrn5]

Thank you @shashwatsai for the detailed explanation of the issue. As per @dbeatty10 and even based on my understanding on DBT architecture, it doesn't allow to manage state of the models in cluster mode kind of job flow. But as i mentioned in my previous replies what if we have remote tracker to track all the state so that we can plan to design some solution for this.

Also, i guess the run_results will be written at the end of the dbt run which will make this more error prone when the client machine goes down in between dbt run and even if we use the dbt retry it wont be able to resume the DAG i guess, instead we can have intermittent checkpointing so that we can address this scenario. But yeah, this is just additional improvement, but main problem is the one we are discussing.

@uplsh580 Thank you for your views and sharing your architecture on this. But i see in your architecture, as you mentioned you are not using any features of dbt(like parallel execution of independent models) which is a major drawback. Even i had a look at astronomer, but it basically breaks down the models to each airflow tasks, and submits as a different spark application.

@dbeatty10 how does it work in dbt cloud with airflow, it launches as a single task in airflow for one complete dbt run? If yes, how the failure case is handled, like lets say few models failed, and even we have the state maintained in run_results.json, but still since its single task, either we have edit the DAG or is any other way or it restarts from beginning?

[uplsh580]

@akashrn5 You're right. I run each model divided into each Airflow task or DAG. This is because among my requirements, I want to control(On, Off and retry) the execution of individual models during repeated scheduling (daily, hourly, weekly, ...).

[JCZuurmond]

Hi all,

TBH I am not completely sure if I understand your needs correctly. I will detail my usual set-up, maybe it could assist your architecture choices.

dbt requires compute (VMs) in two location to:

1. Run dbt:
   • usually via the CLI.
   • usually one VM.
2. Execute the SparkSQL:
   • usually a Spark cluster.
   • usually multiple VMs (driver and executor nodes).

Options for the execution locations I use to:

1. Run dbt:
   1. CI: I used Github Actions or Azure DevOps, any CI suffices
   2. dbt cloud jobs
2. Execute the SparkSQL:
   1. Databricks

Alternatives for the execution locations I would consider:

1. Run dbt:
   1. Airflow: I would investigate this first. One power of dbt lays in the DAG selection. Airflow brings a DAG too, which might make it less clear what should go in which DAG.
   2. Docker container: Run it from anywhere, might make sense to use.
2. Execute the SparkSQL:
   1. Kubernetes
   2. Legacy (on-prem) Spark cluster

What approach to use depends heavily on your context, I would decide based on:

• Current tech stack
• Skills within the team and company
• Functional requirements
• Security and technical requirements. E.g. If a company is strict on networking, you need dbt cloud enterprise; when using a CI you can easily deploy your VM where required.
• Costs

I did not run into the issue of multi-hour dbt executions being cancelled. I expect that I would start with optimizing the execution strategy instead of building a recovery system. Though, I do not know your context and use cases, so that approach might not make sense if I would have a deeper understanding on the context you are operating in.