How are Dagster-Cloud folks managing their costs with long Airbyte syncs? We have a daily ~90 minute sync that makes up the bulk of our cloud

compute minute

costs, but my understanding is that the Dagster job is essentially just monitoring the output of Airbyte logs. Is there way to minimize this to avoid 'double dipping' on compute costs? Is the solution just to schedule those long running jobs natively in Airbyte and declaring them as source assets in Dagster?

Hi, is there a way to integrate airbyte cloud in dagster? Thank you

Seems like a couple people here would be interested. Here's the (experimental) guide on how to connect Airbyte Cloud with Dagster. courtesy of @ben

I raised an issue for the issues with the ingestion as code package since

1.3

, but I think I've identified a fairly easy fix. Essentially, starting with the AirbyteManagedReconciler, simply changing the input parameter to an AirbyteResource, and then removing the param_checks looking for a

ResourceDef

, and doing the same on this class and this function made the whole thing work with the new Airbyte pythonic resource. I just edited my local install of Dagster to test, and was able to get

dagster dev

to run, and the cli tool to deploy config to Airbyte as well. I'm following the contributing instructions to set up a local environment for contributing right now, but I'm a little wary. I'm assuming the tests for this part of the library didn't catch this bug up-front, so I worry my changes might cause some other uncaught regression. Is there anyone more familiar with this part of the code that can help verify my assumptions/changes above?

Hi, I have some issue for connecting airbyte in dagster with error log

ValueError: Airbyte connections are not in sync with provided configuration

. I run it from my docker compose in virtual machine. Any insight from it? I will share my code as well

Hi, I was wondering if there's a way to change the default asset group on airbyte assets. Thank you

Hi all. I am using airbyte cloud and dagster cloud, and I am concerned about the cost if I ran the airbyte resources through dagster cloud and it takes some time to complete. Will dagster incur cost while airbyte is running on the other end? Thanks!

Does anyone else who uses ingestion-as-code have a use case with a set of streams that have an incremental update schedule and a full refresh schedule? We're syncing from Salesforce, and our setup has many formula fields, which slowly drift with incremental syncs, so once a week we'd like to full refresh to bring it up to date. The issue I'm running into is that, if I set up 2 connections with different syncing patterns in Airbyte, they'll show up as separate data assets in Dagster. Is there a neat way to coalesce those, or maybe some alternative solution?

Hey everyone! Just a note that we just launched our no-code connector builder at Airbyte! It now takes about 10 minutes to build a new connector from scratch. No help needed from data engineers. No local development environment required. Here’s a ! Don’t hesitate to give us a shoutout on ProductHunt today!

What are the implications of setting

forward_logs

to False? Is Dagster still aware if the Airbyte sync succeeds or fails? Does it just mean I have to go introspect the error in Airbyte itself if it fails?

To follow up on the above. The documentation says that streaming the logs can get expensive for long running jobs. Does that mean disabling them makes it such that Airbyte syncs aren't accruing dagster cloud compute minutes for the entire duration of the sync any longer?

Hello. I present this problem that I have trying to implement a dagster flow: I'm loading all my airbyte assets with this line of code:

airbyte_assets = load_assets_from_airbyte_instance(airbyte_instance)

Additionally, I have defined another asset that executes a stored procedure in SQL SERVER:

@asset(
        group_name="sql_server_assets",
        retry_policy=RetryPolicy(max_retries=5, delay=60),
)
def sp_sql_server():
    
    # - Credenciales
    server = 'xxx'
    database = 'xxx'
    username = 'xxx'
    password = 'xxx'
    
    # - Cadena de conexión
    connection_string = f"DRIVER={{ODBC Driver 18 for SQL Server}};SERVER={server};DATABASE={database};UID={username};PWD={password}"

    with pyodbc.connect(connection_string) as connection:
        connection.execute("EXEC dbo.InsertTimestamp")

    return None

My problem lies in the need to define dependencies between assets, that is, I want my

sp_sql_server

asset to be executed first and then, through a dependency, a specific asset that is loaded from airbyte is executed, how can I do this? I reviewed the example by benpankow @ben that is on github and there I see a way to define this dependency but in this example it is done the other way around, that is, first the airbyte asset is executed and then the other assets that are defined under the

"non_argument_deps"

configuration "

"depends_on"

. how to achieve something similar but in reverse way? I thank you in advance for any help.

mysql_airbyte_assets = build_airbyte_assets(connection_id=connection_id, destination_tables=["table"])

I have an airbyte connection to do MySQL -> S3 sync. If I materialize above asset, I want to know S3 object path. How do I retrieve that in dagster?

mysql_airbyte_assets = build_airbyte_assets(connection_id=connection_id, destination_tables=["foobar"])
foobar = mysql_airbyte_assets[0]


@asset(ins={"foobar": AssetIn("foobar")})
def save_to_starrocks(context, foobar):
    print("foobar", foobar, context.__dict__)
    return False

Above snippet prints -

foobar None {'_pdb': None, '_events': [], '_output_metadata': {}}

I checked that AirByte materialization is working fine, but can anyone share hints on why

foobar

asset is None here? ^

Hey all. I'm trying to get my job working with partitions. I'm using DailyPartitionsDefinition and everything downstream of the airbyte assets play nice, but if I do a backfill the airbyte sync is called many times leading to a timeout and failure of backfill for the specific day. Am I doing something wrong with the airbyte assets? I'm importing them using load_assets_from_airbyte_instance This is the error:

2023-06-16 13:34:42 +0100 - dagster - ERROR - my_upstream_job - 55a4ed2c-611b-4ccf-a52e-95e607c292aa - airbyte_sync_2db70 - Request to Airbyte API failed: 502 Server Error: Bad Gateway for url: <http://localhost:8000/api/v1/jobs/cancel>
2023-06-16 13:34:42 +0100 - dagster - ERROR - my_upstream_job - 55a4ed2c-611b-4ccf-a52e-95e607c292aa - 6099 - airbyte_sync_2db70 - STEP_FAILURE - Execution of step "airbyte_sync_2db70" failed.

dagster._core.definitions.events.Failure: Max retries (3) exceeded with url: <http://localhost:8000/api/v1/jobs/cancel>.

Hi! I was wondering if you are planning to add support for column level control for each asset similar to what Airbyte recently released: https://airbyte.com/blog/airbyte-column-selection-control-over-the-exact-data-to-sync It would be neat if you could programmatically filter out columns that contain PII for instance.

I'm trying out ingestion as code. I tried to define my MySQL source but got this error. Not sure how to solve this. I've tried

ssl=False

, but it's still required me to give

ssl_mode

. I've tried

ssl_mode="Preferred"

and

ssl_mode=MysqlSource.Preferred

with no luck. Can anyone help? Thanks

dagster._check.ParameterCheckError: Param "ssl_mode" is not one of ['Preferred', 'Required', 'VerifyCA', 'VerifyIdentity']. Got <class 'dagster_airbyte.managed.generated.sources.MysqlSource.Preferred'> which is type <class 'type'>.

Is there a way to set the destination table name from stream_config?

Hi, Do you plan to implement the automaterialize policy for Airbyte Cloud?

Hey everyone! 👋 Just some big news to share on Airbyte’s side: our official Airbyte API and *Terraform Provider are both now available on Airbyte Open Source!* Here’s a more detailed article about the release.

<solved, stupid question>

Hey there! Is there any reason why

build_airbyte_assets

does not accept an

io_manager_key

parameter while

load_assets_from_airbyte_project

does ? (https://docs.dagster.io/_apidocs/libraries/dagster-airbyte#dagster_airbyte.build_airbyte_assets)

I'm trying to upgrade from

1.4.1

to

1.4.5

and I am getting the error message

dagster._core.errors.DagsterInvalidDefinitionError: Conflicting versions of resource with key 'airbyte' were provided to different assets. When constructing a job, all resource definitions provided to assets must match by reference equality for a given key.

I'm passing the same instance of

AirbyteResource

to all of my

load_assets_from_airbyte_instance

calls. Does anyone have any insight here?

I am using Airbyte to build integrations for our customers for some specific source systems. As such, I will likely have many stream with exactly the same name (but will live in a different folder in S3, my destination). Is there a way in Dagster to "prefix" the assets that are automatically loaded from

load_assets_from_airbyte_instance

by the actually connection id?

Hey. I have a bunch of Airbyte streams, all going to S3, prefixed w/

$NAME/

(note the

/

), so the stream name winds up being something like

e2e/data

. When I use

construct_airbyte_assets

and set _`destination_tables`_ as just

data

, it'll fail with

DagsterStepOutputNotFoundError

. If set

destination_tables

to

e2e/data

, it'll fail with

"e2e/data" is not a valid name in Dagster. Names must be in regex ^[A-Za-z0-9_]+$

. The _`asset_key_prefix`_ only prefixes the asset w/in Dagster, but still doesn't find any valid output from the source. What am I missing - is there not a way to map the source Airbyte stream name into a valid Dagster asset name?

👋 Am wondering if anyone know of a sensible way to define upstream dbt assets for airbyte assets? we have some airbyte connections that are depending on dbt assets, and would like to link them so things like auto-materialization would work, i.e. once a dbt asset is materialized, the system will kick of an airbyte connection. I’m loading my airbyte connections via instance integration (

load_assets_from_airbyte_instance

). What would be nice if i dont have to define them by hand one by one… Thanks!

Hi, I have an upstream DBT asset for my airbyte asset. I have defined the airbyte asset using

build_airbyte_assets

. The airbyte connection extracts data from snowflake and outputs a csv file into an azure blob container. The pipeline is working, in that it outputs the file, however the airbyte asset fails with the following error:

op 'airbyte_sync_820ee' did not fire output {bsv}

dagster._core.errors.DagsterStepOutputNotFoundError: Core compute for op "airbyte_sync_820ee" did not return an output for non-optional output "bsv"

Any ideas on how to resolve this? how could I configure this asset to not require an output? Thanks

Hi, there seems to be a discrepancy in the DatabricksDestination. Per Airbyte doc, the data source can be DB managed tables, or 2 external Blob storages. When I try the managed tables option in the UI, the destination is created successfully. However, when I do it in the dagster-airbyte library, only 2 external Blob storages are allowed as data source (see error below). The source code does reflect that.

dagster._check.ParameterCheckError: Param "data_source" is not one of ['AmazonS3', 'AzureBlobStorage']. Got 'MANAGED_TABLES_STORAGE' which is type <class 'str'>.

Hello, I am using Airbyte to bring in data from 2 db's into S3. These 2 db's are similar however are being used by different teams within the organization and I am trying to join the data I am testing with 1 table at the moment (there are several tables i plan to bring in). When I bring in the Airbyte asset into Dagster I am getting duplicate key error.

dagster._core.errors.DagsterInvalidDefinitionError: Duplicate asset key: AssetKey(['table_key'])

Is there a way to differentiate between the the 2 different asset keys between the 2 db's? maybe added a tag or pull the connection or source name as part of the asset key?

Hello everyone, When I am loading multiple assets using

load_assets_from_airbyte_instance

with normalisation on airbyte being set to `off`: • the loaded assets come with names like

{the_prefix_I've_set}/{airbyte_stream_name}

. • The tables that are created on BigQuery have names like

{the_prefix_I've_set}._airbyte_raw_{airbyte_stream_name}

• The problem is that I struggle to create proper data lineage and dependencies when I later use DBT on a table named

{the_prefix_I've_set}._airbyte_raw_{airbyte_stream_name}

that doesn’t exist for Dagster. Is there any way for the assets to follow the names of the created from airbyte tables when automatically loaded? Any other workarounds?