How do I set my DbtCliProfile to use a SnowflakeTargetConfig? The UI only let’s you select TargetConfig and not a snowflake one

Just researching: • Does anyone have documentation or references on maxing out dbt dbt CORE capabilities before migrating to Prefect?

Hi all! I'm trying to run a dbt library locally using dbt-cli in Prefect. I've noticed an error I tend to get is:

ERROR: Runtime Error
  Credentials in profile "snowflake_dbt_transformations", target "dev" invalid: 'database' is a required propertyDefined profiles:
 - snowflake_dbt_transformations
For more information on configuring profiles, please consult the dbt docs:

<https://docs.getdbt.com/docs/configure-your-profile>

Upon investigation, it seems like the dbtCliProfile passed to the dbtTrigger task gets saved in the

~/.dbt/profiles.yml

first and then used inside the dbtCli shell task. However the problem is the method where it saves the dbtCliProfile into a yaml file does not save the

TargetConfigs

resulting in the profile file being incomplete. Has anyone had an experience with this before and know any workarounds?

Hi! I'm standing up dbt CLI on AWS Fargate using tasks from this template: https://github.com/anna-geller/dataflow-ops I have all of the flows deployed and scheduled using Prefect cloud, and I'm having no problems running the dbt run (it is using the project files from the S3 bucket/folder that's attached to the image), but I'm running into issues when running

dbt docs generate

as the Flow Run is outputting that

Catalog written to /opt/prefect/target/catalog.json

I checked S3 and it's definitely not there, and I'm guessing that the

/opt

is a directory that's tied to the ECS Task from the container? Is there a way that the dbt cli command can output the docs file (specifically index.html into the S3 bucket so we can host our dbt docs as a static site) Flow code is in the 🧵. I really appreciate any help or pointers 🙂

Hi everyone, please how can I run prefect 2.0 Agent within python code that contain the flow?

Hi, this might be a stupid question, but is it possible to schedule queries created with dbt inside Bigquery? We are looking to have views, ... created into Bigquery and would like to know if I can schedule when they run inside a flow. Atm we have them as scheduled queries that run at specific times. Thanks!

Cross posting the announcement of prefect-dbt v0.3.0 https://prefect-community.slack.com/archives/CKNSX5WG3/p1676578412722889

Hi all! Very much starting out trying to orchestrate our dbt commands using prefect but running into a few issues. Issues in images provided - can anyone guide me on why these flows aren't working? First image - no idea why it doesn't like that comma as I've lifted that snippet from prefect-dbt github page. Second image - I'm running a task from within a flow, but it seems to this I'm running a task from within a task?

Hello all, I would like to ask for your recommendations on structuring flows/subflows/tasks. We would like to define as many tasks as possible, so that the radar displays the entire job as meaningfully as possible (without having to jump from flow to flow or subflow; instead, we want to see an overall picture). We run into the problem that dbt_cli itself is also a task (correct me if I'm wrong). What are your recommendations here? How do you proceed with the interaction of Prefect and dbt? Many thanks in advance @Florian Pape, fyi

Hi all, Anyone have any idea where I'm going wrong here? Just can't seem to get this working? Keep getting this error message.

X-posting, see you all in #data-ecosystem

Hi everyone, has anyone of you ever had experience with the following error? The error occurs when executing a prefect flow that runs multiple dbt jobs. We host Prefect ourselves via Docker/Kubernetes. Do you have any ideas on the cause and remedy?

RuntimeError: Command failed with exit code 133:
/tmp/prefect-4zqou1u7: line 1: 299284 Trace/breakpoint trap   (core dumped) dbt build --select dl_nicc_ecom__order_header --profiles-dir /tmp/dbt --project-dir /tmp/dbt
12:14:43 PM
Finished in state Failed('Flow run encountered an exception. RuntimeError: Command failed with exit code 133:\n/tmp/prefect-4zqou1u7: line 1: 299284 Trace/breakpoint trap   (core dumped) dbt build --select dl_nicc_ecom__order_header --profiles-dir /tmp/dbt --project-dir /tmp/dbt\n\n')

Many thanks in advance!

Anyone able to support me here? 🙂

When attempting to use

DbtCoreOperation

we are receiving the following error... This suggests that our dbt cli profile block, which we have defined in dbt Cloud, does not have a name or target or target_configs, but it does..

pydantic.error_wrappers.ValidationError: 3 validation errors for DbtCoreOperation
dbt_cli_profile -> name
  field required (type=value_error.missing)
dbt_cli_profile -> target
  field required (type=value_error.missing)
dbt_cli_profile -> target_configs
  field required (type=value_error.missing)

I was able to use the

profile_dir

parameter to use my local credentials, but this only verifies there aren't obvious problems with the flow code.

prefect-dbt==0.3.1

prefect-snowflake==0.26.0

Any ideas?

More difficulties with

dbtCoreOperation

. I am running...

with DbtCoreOperation(
        commands=["dbt deps", "dbt build " + dbt_build_arg + " --target " + DBT_ENV],
        project_dir="edw-dbt",
        # working_directory="edw-dbt",
        # overwrite_profiles=True,
        profiles_dir="~/.dbt",
        # dbt_cli_profile=dbt_cli_profile,
        stream_output=True,
    ) as dbt_operation:
        dbt_process = dbt_operation.trigger()
        dbt_process.wait_for_completion()
        result = dbt_process.fetch_result()

Currently failing with:

15:35:09.705 | INFO    | Flow run 'furry-carp' - PID 98611 stream output:
09:50:09  Error importing adapter: No module named 'dbt.adapters.snowflake'
15:35:09.706 | INFO    | Flow run 'furry-carp' - PID 98611 stream output:
09:50:09  Encountered an error while reading profiles:
  ERROR: Runtime Error
  Credentials in profile "edw", target "dev" invalid: Runtime Error
    Could not find adapter type snowflake!Defined profiles:
 - edw
For more information on configuring profiles, please consult the dbt docs:

<https://docs.getdbt.com/docs/configure-your-profile>
15:35:09.720 | INFO    | Flow run 'furry-carp' - PID 98611 stream output:
09:50:09  Encountered an error:
Runtime Error
  Could not run dbt

Hi EveryOne ! I'm getting this error with dbtCoreOperation , can anyone tell me what this error means?

Finished in state Failed('Flow run encountered an exception. RuntimeError: PID 17072 failed with return code 1.\n')

Hey folks 👋 Are there any example on how to trigger dbt core on a Redshift target? Looking at

prefect-dbt

I see the available targets are Snowflake, Bigquery and Postgres, but not Redshift. Should I use the Postgres target? Any hint is much appreciated! 🙌

Hi there! Are there any example on how to configure a dbt profile on an Athena target? I'm having some trouble to find infos for generic connection

Hi everyone! I have been loving Prefect. I am working on adding dbt to my pipelines and after reading the docs and encountering errors in a few different approaches I thought I would come here for help. Adding code in the comments to keep the size down. Thanks in advance!

Hi everyone! Any idea about this issue? At least from Prefect 2.10.0, Prefect is not usable

Hi! I am new to Prefect and am trying to get DBT flows up and running in my local prefect `2.10.12`/Python

3.10

environment and am hitting some errors. Is there anyone who can take a look and let me know if I'm doing something wrong? Going to put some more info in the thread to not clog up the channel..

[Local prefect][dbt core][Windows OS] Hi all, I tried adding dbt to my pipeline, but encounter an error saying:

RuntimeError: PID 29636 failed with return code 1.

and with another section saying:

OneDrive\Documents\WindowsPowerShell\profile.ps1 cannot be loaded because running scripts is disabled  
on this system. For more information, see about_Execution_Policies at https:/go.microsoft.com/fwlink/?LinkID=135170.

My code and more info in this thread:

Wondering if anyone has run into something like this error before, and if so, how you resolved it:

File "/usr/lib/python3.10/subprocess.py", line 1845, in _execute_child
    raise child_exception_type(errno_num, err_msg, err_filename)
 FileNotFoundError: [Errno 2] No such file or directory: 'bash'

I'm hitting this error any time that I try to trigger a DbtCoreOperation from a deployment. I'm running the most recent version of the prefect server and task runner, and most recent versions of the prefect_dbt and prefect_airbyte pip packages. This is running on an Ubuntu 22.04 server. It's using the default local file storage for the flow, and a local agent running on the same server as the Prefect server process. I'm hitting the above exception as soon as I try to kick off any DbtCoreOperation, as soon as either "run()" or "trigger()" is called. The workflow runs perfectly when running the script manually, but fails on the dbt stuff as soon as I try running it through a deployment. The Airbyte operation runs fine from both environments, so this is specific to the dbt operations. Any help would be much appreciated! Code below:

import sys
from prefect import flow
from prefect_dbt.cli import DbtCoreOperation
from prefect_airbyte.server import AirbyteServer
from prefect_airbyte.connections import AirbyteConnection


DBT_PROJECT_DIR = "<path to dbt project>"
DBT_PROFILE_DIR = "<path to dbt project>"
AIRBYTE_CONNECTION = "<connection id>"

airbyte_server = AirbyteServer(server_host="<host>", server_port=<port>)

@flow(log_prints=True)
def dbt_snapshot():
    with DbtCoreOperation(commands=["dbt snapshot"],
                project_dir = DBT_PROJECT_DIR,
                profiles_dir = DBT_PROFILE_DIR) as dbt_run:
        
        dbt_process = dbt_run.trigger()
        dbt_process.wait_for_completion()
        dbt_output = dbt_process.fetch_result()
        return dbt_output

@flow(log_prints=True)
def dbt_model():
    result = DbtCoreOperation(
        commands = [f"dbt run --select models/*"],
        project_dir = DBT_PROJECT_DIR,
        profiles_dir = DBT_PROFILE_DIR
    ).run() 
    return result

@flow(log_prints=True)
def airbyte_sync():
      connection = AirbyteConnection(
            connection_id = AIRBYTE_CONNECTION,
            airbyte_server = airbyte_server
      )
      job_run = connection.trigger()
      job_run.wait_for_completion()
      res = job_run.fetch_result()
      return res

@flow(log_prints=True)
def dbt_airbyte_sync():
    dbt_snapshot()
    dbt_model()
    airbyte_sync()

if __name__ == "__main__":
    try:
        dbt_airbyte_sync()
    except Exception as err:
        print("Exception while running workflow", err)
        sys.exit(1)
    sys.exit(0)

I'm trying to implement a dbt Cloud Credentials block but I'm getting an error. My code:

from prefect_dbt.cloud import DbtCloudCredentials

dbt_cloud_credentials = DbtCloudCredentials.load("dbt-cloud")

Subject: Request for Assistance with Task Naming in Prefect UI for dbt Models Hello everyone👋, hope you are doing well! I am writing to discuss an issue we are experiencing with our use of Prefect and dbt in our current project. Within our Prefect flow, we utilize multiple tasks through the

trigger_dbt_cli_command

, each performing various CLI commands. More specifically, we run different dbt models based on certain tags. The issue we encounter is within the Prefect UI, where these dbt tasks are automatically assigned default names (such as

trigger_dbt_cli_command-0

,

trigger_dbt_cli_command-1

, etc.). This automatic naming becomes problematic when a task fails, as it is difficult to discern precisely which task has encountered the error. From the docs I didn't manage to find any solution to that. We would greatly appreciate your insight and any potential solutions to this naming issue. 😁

any news about this issue? Basically we cannot upgrade because of this

when I install prefect-dbt-0.3.1, prefect 2.11.1 get uninstalled and replaced by prefect 2.7.7

Hello. Referencing the following page: https://prefecthq.github.io/prefect-dbt/#__code_12 When I make changes to the code placeholders according to the information on the above page and execute the Python code, it results in an error. If anyone knows a solution, please share it. Version:

Python Version: 3.11.4
prefect-dbt Version: 0.3.1

Code:

from prefect_dbt.cloud import DbtCloudCredentials, DbtCloudJob

dbt_cloud_credentials = DbtCloudCredentials.load("BLOCK-NAME-PLACEHOLDER")
dbt_cloud_job = DbtCloudJob.load(
    dbt_cloud_credentials=dbt_cloud_credentials,
    job_id="JOB-ID-PLACEHOLDER"
)

Error Message:

$ python create_job_block.py
Traceback (most recent call last):
  File "/Users/k/work/prefect/dbt-snowflake/create_job_block.py", line 4, in <module>
    dbt_cloud_job = DbtCloudJob.load(
                    ^^^^^^^^^^^^^^^^^
  File "/usr/local/Caskroom/miniconda/base/lib/python3.11/site-packages/prefect/utilities/asyncutils.py", line 255, in coroutine_wrapper
    return call()
           ^^^^^^
  File "/usr/local/Caskroom/miniconda/base/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 383, in __call__
    return self.result()
           ^^^^^^^^^^^^^
  File "/usr/local/Caskroom/miniconda/base/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 283, in result
    return self.future.result(timeout=timeout)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/Caskroom/miniconda/base/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 169, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/Caskroom/miniconda/base/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/usr/local/Caskroom/miniconda/base/lib/python3.11/site-packages/prefect/_internal/concurrency/calls.py", line 346, in _run_async
    result = await coro
             ^^^^^^^^^^
  File "/usr/local/Caskroom/miniconda/base/lib/python3.11/site-packages/prefect/client/utilities.py", line 51, in with_injected_client
    return await fn(*args, **kwargs)
                 ^^^^^^^^^^^^^^^^^^^
TypeError: Block.load() got an unexpected keyword argument 'dbt_cloud_credentials'
$

Hi team, I’m setting flows for my dbt core on BigQuery following prefect-dbt docs. It appears to me that I need to declare a specific dataset for each

BigQueryTargetConfig

block in order to register a

DbtCliProfile

block ? Let’s say I have 30 datasets on prod, do I have to do this step 30 times? This sounds quite cumbersome for me so I’d appreciate if you could give me any pointer on this

from prefect_gcp.credentials import GcpCredentials
from prefect_dbt.cli import BigQueryTargetConfigs, DbtCliProfile

credentials = GcpCredentials.load("CREDENTIALS-BLOCK-NAME-PLACEHOLDER")
target_configs = BigQueryTargetConfigs(
    schema="SCHEMA-NAME-PLACEHOLDER",  # also known as dataset
    credentials=credentials,
)
target_configs.save("TARGET-CONFIGS-BLOCK-NAME-PLACEHOLDER")

dbt_cli_profile = DbtCliProfile(
    name="PROFILE-NAME-PLACEHOLDER",
    target="TARGET-NAME-placeholder",
    target_configs=target_configs,
)
dbt_cli_profile.save("DBT-CLI-PROFILE-BLOCK-NAME-PLACEHOLDER")

Hi! I wanna get an error from dbt and send to my task from prefect. how could i do that ?