Hi Do you plan to implement the automaterialize policy for A dagster #integration-airbyte

Join Slack

Hi, Do you plan to implement the automaterialize p...

# integration-airbyte

Julien DEBLANDER

08/01/2023, 4:19 PM

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

owen

08/01/2023, 6:39 PM

hi @Julien DEBLANDER -- the

load_assets_from_airbyte_instance

(and related) functions support a

connection_to_auto_materialize_policy_fn

, which might be what you're looking for here

Julien DEBLANDER

08/01/2023, 8:28 PM

Is it supported with the latest Dagster version? I'm currently on 1.3.12 and I get the error: "load_assets_from_airbyte_instance is not yet supported for AirbyteCloudResource"

owen

08/01/2023, 8:33 PM

Ah sorry about that, I somehow missed that this was referring to Airbyte Cloud. This is not supported at the moment, as the Airbyte Cloud API doesn't currently support all of the endpoints that we use in the integration. What is your current setup with Dagster/Airbyte Cloud?

Julien DEBLANDER

08/01/2023, 8:40 PM

I'm just at the beginning of the implementation of Airbyte Cloud but we are currently fetching data with Airbyte on a schedule. Then I transform this data with dbt.

Julien DEBLANDER

08/01/2023, 8:41 PM

I guess I should stick to the schedule definition for now 🙂

owen

08/01/2023, 8:49 PM

got it, that sounds right to me

Shangwei Wang

08/14/2023, 9:55 PM

👋 am interested in setting up AMP for airbyte here too. We are hosting Airbyte ourselves in our own cluster, based on this thread, am i correct to assume

connection_to_auto_materialize_policy_fn

should work with OSS airbyte? I have AMP setup for all my dbt models, and Airbyte assets (defined as dbt

source

for my dbt models to get that nice lineage) via

load_assets_from_airbyte_instance

, but AMP doens’t show up for airbyte assets… since AMP doens’t show up, my dbt models are not being auto-materialized until I manually materialize the Airbyte assets… any pointers appreciated, thanks! (I’m on dagster 1.4.5)

owen

08/15/2023, 4:52 PM

Hi! I would expect that to work at the moment -- mind sharing the code you're using to set that up? If all else fails, it's possible to manually add AMPs to assets using something like

Copy code

list_of_assets = [...] # get this from load_assets_from_airbyte_instance
modified_list_of_assets = [
    assets_def.with_attributes(auto_materialize_policy=...) for assets_def in list_of_assets
]

Shangwei Wang

08/15/2023, 4:54 PM

Copy code

airbyte_assets = load_assets_from_airbyte_instance(  
      airbyte.airbyte_instance,
      key_prefix=["airbyte"],
      connection_to_group_fn=
          lambda connection_name:
              f'airbyte_{re.sub("[^0-9a-zA-Z_]+", "_", connection_name)}'.lower(),
      connection_filter=lambda connection: connection.name.startswith("Warehouse"),
      connection_to_auto_materialize_policy_fn=lambda _: AutoMaterializePolicy.lazy(),
  )

Shangwei Wang

08/15/2023, 4:54 PM

thanks, here is the init code

owen

08/15/2023, 10:13 PM

got it, not sure why that's happening -- but just to help debug, if you try out the

with_attributes

method above, does that resolve the issue?

Shangwei Wang

08/15/2023, 11:11 PM

will give that a try and report back, thanks!

Shangwei Wang

08/16/2023, 10:48 PM

just realized that

auto_materialize_policy

isn’t a supported param for

with_attributes

Shangwei Wang

08/16/2023, 10:56 PM

BUT, looks like the following works, now my airbyte jobs are able to auto-materialize accordingly to downstream dbt freshness policy 🎉

Copy code

airbyte_assets = _airbyte_assets.with_attributes_for_all(
     group_name=None,
     freshness_policy=None,
    backfill_policy=None,
     auto_materialize_policy=AutoMaterializePolicy.lazy(),
  )

🌈 1

Open in Slack

Previous Next