hey, I have some issues with

dagster-airbyte

integration plus some feature ideas (what I can add/contribute). is here anyone who is aware of the airbyte integration? I’d explain it in details what hurts me

@sandy who else is familiar with the airbyte integration with owen OOO?

I'd also make a contribution if possible, making the table list optional in build_airbyte_assets. if that parameter is empty dagster should refresh all tables in the specified connection

while these tables were updated by airflow as part of the built-in transformation stage, I cannot tell dagster to assume those tables refreshed.

do you mind clarifying what you mean by this a little further?

OK, let me explain this buit-in transformations:

that makes sense. can I ask you to try something? instead of using the

build_airbyte_assets

in the dagster-airbyte library, define your own:

from typing import List, Optional

from dagster_airbyte.utils import generate_materializations

from dagster import AssetKey, AssetOut, Output
from dagster import _check as check
from dagster._annotations import experimental
from dagster._core.definitions import AssetsDefinition, multi_asset


@experimental
def build_airbyte_assets(
    connection_id: str,
    destination_tables: List[str],
    asset_key_prefix: Optional[List[str]] = None,
) -> List[AssetsDefinition]:
    """
    Builds a set of assets representing the tables created by an Airbyte sync operation.

    Args:
        connection_id (str): The Airbyte Connection ID that this op will sync. You can retrieve this
            value from the "Connections" tab of a given connector in the Airbyte UI.
        destination_tables (List[str]): The names of the tables that you want to be represented
            in the Dagster asset graph for this sync. This will generally map to the name of the
            stream in Airbyte, unless a stream prefix has been specified in Airbyte.
        asset_key_prefix (Optional[List[str]]): A prefix for the asset keys inside this asset.
            If left blank, assets will have a key of `AssetKey([table_name])`.
    """

    asset_key_prefix = check.opt_list_param(asset_key_prefix, "asset_key_prefix", of_type=str)

    @multi_asset(
        name=f"airbyte_sync_{connection_id[:5]}",
        outs={
            table: AssetOut(key=AssetKey(asset_key_prefix + [table], is_required=False))
            for table in destination_tables
        },
        required_resource_keys={"airbyte"},
        compute_kind="airbyte",
    )
    def _assets(context):
        ab_output = context.resources.airbyte.sync_and_poll(connection_id=connection_id)
        for materialization in generate_materializations(ab_output, asset_key_prefix):
            table_name = materialization.asset_key.path[-1]
            if table_name in destination_tables:
                yield Output(
                    value=None,
                    output_name=table_name,
                    metadata={
                        entry.label: entry.entry_data for entry in materialization.metadata_entries
                    },
                )
            else:
                yield materialization

    return [_assets]

this is exactly the same as the one in the dagster-airbyte library, except it marks the outputs as non-required if this works, we can make this change in the dagster-airbyte library itself

thanks for getting back this early! so, I will try out in a few mins and let you know

are you able to help me understand a little better why we're not registering that

src_tasks_customfields

is being updated? is there some way of detecting this with Airbyte's APIs?

I did not find any APIs, just mentioned in the text logs

hey Tamas - our dagster-airbyte expert (Owen) is out on vacation this week, but will be back next week on Tuesday. I'm going to ask him to look in to this. feel free to ping us if we don't get to it after then

sounds good, thanks!

hi @Tamas Foldi! looked into this a bit, and this is definitely not ideal behavior, but it's a tricky problem to solve, especially because airbyte does not provide a way to query what tables will be produced by a given normalization operation (nor does it provide a way to query what tables were produced by the operation after the fact). It's definitely possible to come up with algorithms that do a pretty good job at guessing output table names based off of the connection schema, but there are definitely edge cases (see: https://docs.airbyte.com/understanding-airbyte/basic-normalization/#naming-limitations--truncation), which I don't think we would want to maintain handling for. For these cases, maybe the conceptually simplest (if somewhat annoying) solution would be to have a separate

destination_table_to_unnested_tables

mapping parameter (bad name, but that sort of idea), which maps the base table to a set of transformed, unnested tables. so for your case that would look like

{
    "src_tasks": {"src_tasks_customfields"}, 
    "src_workflows": {"src_workflows_customstatuses"},
}

then adding those values as additional

destination_tables

. Finally, you could change the logic inside the asset to look like

def _assets(context):
        ab_output = context.resources.airbyte.sync_and_poll(connection_id=connection_id)
        for materialization in generate_materializations(ab_output, asset_key_prefix):
            table_name = materialization.asset_key.path[-1]
            if table_name in destination_tables:
                yield Output(
                    value=None,
                    output_name=table_name,
                    metadata={
                        entry.label: entry.entry_data for entry in materialization.metadata_entries
                    },
                )
                for unnested_table_name in destination_tables_to_unnested_tables.get(table_name, []):
                    yield Output(value=None, output_name=unnested_table_name)
            else:
                yield materialization

yes, actually I already implemented something similar a few days back https://github.com/dagster-io/dagster/pull/9561/files

yeah your solution seems totally reasonable, and if that feels more ergonomic there's definitely nothing wrong w/ it. re: the logs issue, you may need to set up a compute log manager to get the unstructured logs in from stdout/stderr (https://docs.dagster.io/deployment/dagster-instance#compute-log-storage)

thanks. regarding next steps, if adding tests and documentation any chances it can be merged to the main code base?

hi @Tamas Foldi! We're actually looking to include this functionality in next week's release (pr: https://github.com/dagster-io/dagster/pull/9655/files)

looks awesome, really appreciate it

we're adding a general utility function

with_group

in this week's release, so you'll be able to do

with_group(build_airbyte_assets(...), "mygroup")

😄

you guys are good in mind reading

Piggybacking this thread because it seems like it's closely related. I'm getting the "not fired" error on Airbyte, even though the materialization works and the names seem to match my dbt setup. Airbyte is using raw JSON output. Got a very similar issue with s3 as an output as well. The Materializaition looks right, the Airbyte returns successfully, and it is updating in the destination database. When I run the DBT alone it also executes

hey @Bit Unwise! My guess is that you're calling something to the effect of

build_airbyte_assets(connection_id=..., destination_tables=["volleyball_volleyball"], asset_key_prefix=["airbyte_volleyball_volleyball"])

, is that accurate? If so, I think you'd want to swap out

destination_tables=["airbyte_volleyball_details"]

I see, so I need to tweak e.g. DBT for the downstream pieces? I had it listed like this:

version: 2

sources:
  # VOLLEYBALL
  - name: airbyte_volleyball_volleyball
    schema: airbyte_volleyball_volleyball
    tables:
      - name: volleyball_volleyball
        identifier: _airbyte_raw_airbyte_volleyball_details

in

sources.yml

tweaking the dbt source name to match what airbyte's generating (so table name = airbyte_volleball_details") would probably be the most straightforward route. alternatively, you can customize how dagster generates asset keys from your sources.yaml file using the node_info_to_asset_key parameter, if you'd prefer not to have to edit your dbt project

strange. same error for

airbyte_volleyball_details

(the name of the stream) and

_airbyte_raw_airbyte_volleyball_details

(the final output table). Airbyte logs, in case I'm just doing something silly -- quite likely (

op 'airbyte_sync_49b83' did not fire outputs {'airbyte_volleball_details'}

)

2022-09-14 22:26:54 destination > 2022-09-14 22:26:54 INFO i.a.i.d.r.InMemoryRecordBufferingStrategy(lambda$flushAll$1):86 - Flushing airbyte_volleyball_details: 34 records (1 MB)
2022-09-14 22:26:54 destination > 2022-09-14 22:26:54 INFO i.a.i.d.j.SqlOperations(onDestinationCloseOperations):137 - No onDestinationCloseOperations required for this destination.
2022-09-14 22:26:54 destination > 2022-09-14 22:26:54 INFO i.a.i.d.j.JdbcBufferedConsumerFactory(lambda$onCloseFunction$3):162 - Finalizing tables in destination started for 1 streams
2022-09-14 22:26:54 destination > 2022-09-14 22:26:54 INFO i.a.i.d.j.JdbcBufferedConsumerFactory(lambda$onCloseFunction$3):167 - Finalizing stream airbyte_volleyball_details. schema airbyte_volleyball_volleyball, tmp table _airbyte_tmp_fnm_airbyte_volleyball_details, final table _airbyte_raw_airbyte_volleyball_details
2022-09-14 22:26:54 destination > 2022-09-14 22:26:54 INFO i.a.i.d.j.JdbcBufferedConsumerFactory(lambda$onCloseFunction$3):180 - Executing finalization of tables.
2022-09-14 22:26:54 destination > 2022-09-14 22:26:54 INFO i.a.i.d.j.JdbcBufferedConsumerFactory(lambda$onCloseFunction$3):182 - Finalizing tables in destination completed.
2022-09-14 22:26:54 destination > 2022-09-14 22:26:54 INFO i.a.i.d.j.JdbcBufferedConsumerFactory(lambda$onCloseFunction$3):185 - Cleaning tmp tables in destination started for 1 streams
2022-09-14 22:26:54 destination > 2022-09-14 22:26:54 INFO i.a.i.d.j.JdbcBufferedConsumerFactory(lambda$onCloseFunction$3):189 - Cleaning tmp table in destination started for stream airbyte_volleyball_details. schema airbyte_volleyball_volleyball, tmp table name: _airbyte_tmp_fnm_airbyte_volleyball_details
2022-09-14 22:26:54 destination > 2022-09-14 22:26:54 INFO i.a.i.d.j.JdbcBufferedConsumerFactory(lambda$onCloseFunction$3):194 - Cleaning tmp tables in destination completed.
2022-09-14 22:26:54 destination > 2022-09-14 22:26:54 INFO i.a.i.b.IntegrationRunner(runInternal):171 - Completed integration: io.airbyte.integrations.base.ssh.SshWrappedDestination
2022-09-14 22:26:54 destination > 2022-09-14 22:26:54 INFO i.a.i.d.p.PostgresDestination(main):92 - completed destination: class io.airbyte.integrations.destination.postgres.PostgresDestination

"airbyte_volleball_details" -> "airbyte_volleyball_details" (I think that was originally my typo, sorry blob smile sweat )

goodness gracious. Turns out I just can't type, and it works fine when using the

Stream Name

. Super helpful, thank you!

glad it's working!