@dlt_assets decorator and DagsterDltResource instance

[bruk0]

I have a question about difference of @dlt_assets and instance of DagsterDltResource .

I don't know well how they work under the hood . I have an asset which accepts @dlt_assets decorator. In body of function, I can yield the dlt_resource.run() and pass the same parameters I passed into @dlt_assets decorator.

I prefer to use dlt_resource.run(), because there I can provide necessary parameters to dlt Source and Pipeline. But it seems the @dlt_assets part launches.

If I replace decorator to @asset , the part of dlt_resource.run() , during a run I get an error that the asset is not found in definitions. So I think, it's a problem related to multi_asset mechanism.

I am looking for a way to pass parameters to @dlt_assets dynamically from the upstream asset. Currently, it only works if statically define parameters.

Thanks for your time to read this!

The code snippet with both DagsterDltResource and @dlt_assets:

@dlt_assets(
    dlt_source=google_analytics(
        property_id=config.property_id,
        start_date=config.start_date,
        queries=config.queries,
    ),
    dlt_pipeline=pipeline(
        pipeline_name=config.pipeline_name,
        destination='bigquery',
        dataset_name=config.dataset_name,
    ),
    name='google_analytics_asset',
    group_name='Google_Analytics',
    dlt_dagster_translator=CustomDagsterDltTranslator()
)
def google_analytics_asset(context: AssetExecutionContext, dlt_resource: DagsterDltResource, config: PipelineConfig, source_secret, destination_secret):
    context.log.info(config.start_date)
    context.log.info(config.queries)
    context.log.info(config.property_id)
    context.log.info(config.organization_id)
    context.log.info(type(source_secret))
    context.log.info(source_secret)
    context.log.info(destination_secret)

    dlt.secrets[f'{config.pipeline_name}.sources.google_analytics.credentials.client_id'] \
        = source_secret[0]
    dlt.secrets[f'{config.pipeline_name}.sources.google_analytics.credentials.client_secret'] \
        = source_secret[1]
    dlt.secrets[f"{config.pipeline_name}.sources.google_analytics.credentials.refresh_token"] \
        = source_secret[2]
    dlt.secrets[f'{config.pipeline_name}.sources.google_analytics.credentials.project_id'] \
        = config.project_id

    os.environ[f'{config.pipeline_name}__SOURCES__GOOGLE_ANALYTICS__CREDENTIALS__CLIENT_ID']=source_secret[0]
    os.environ[f'{config.pipeline_name}__SOURCES__GOOGLE_ANALYTICS__CREDENTIALS__CLIENT_SECRET']=source_secret[1]
    os.environ[f'{config.pipeline_name}__SOURCES__GOOGLE_ANALYTICS__CREDENTIALS__REFRESH_TOKEN']=source_secret[2]
    os.environ[f'{config.pipeline_name}__SOURCES__GOOGLE_ANALYTICS__CREDENTIALS__PROJECT_ID']=config.project_id

    client_id = dlt.secrets[f'{config.pipeline_name}.sources.google_analytics.credentials.client_id'] = source_secret[0]
    context.log.error(f"CLIENT_ID FROM SOURCE SECRET ASSET: {client_id}")
    client_secret = dlt.secrets[f'{config.pipeline_name}.sources.google_analytics.credentials.client_secret'] = source_secret[1]
    context.log.error(f"CLIENT_SECRET FROM SOURCE SECRET ASSET: {client_secret}")
    refresh_token = dlt.secrets[f"{config.pipeline_name}.sources.google_analytics.credentials.refresh_token"] = source_secret[2]
    context.log.error(f"REFRESH TOKEN FROM SOURCE SECRET ASSET: {refresh_token}")
    context.log.error(f"PROJECT_ID FROM SOURCE SECRET ASSET: {config.project_id}")

    org_client_email = dlt.secrets[f'{config.pipeline_name}.destination.bigquery.credentials.client_email'] \
        = destination_secret[2]
    org_private_key = dlt.secrets[f'{config.pipeline_name}.destination.bigquery.credentials.private_key'] \
        =  destination_secret[1]
    org_project_id = dlt.secrets[f'{config.pipeline_name}.destination.bigquery.credentials.project_id'] \
        = destination_secret[0]

    os.environ[f"{config.pipeline_name}__DESTINATION__CREDENTIALS__CLIENT_EMAIL"] = destination_secret[2]
    os.environ[f"{config.pipeline_name}__DESTINATION__CREDENTIALS__PRIVATE_KEY"] = destination_secret[1]
    os.environ[f"{config.pipeline_name}__DESTINATION__CREDENTIALS__PROJECT_ID"] = destination_secret[0]

    context.log.warning(f"{config.pipeline_name}__DESTINATION__CREDENTIALS__CLIENT_EMAIL")
    context.log.warning(f"{config.pipeline_name}__DESTINATION__CREDENTIALS__PRIVATE_KEY")
    context.log.warning(f"{config.pipeline_name}__DESTINATION__CREDENTIALS__PROJECT_ID")


    context.log.warning(os.environ.get(f"{config.pipeline_name}__DESTINATION__CREDENTIALS__CLIENT_EMAIL"))
    context.log.warning(os.environ.get(f"{config.pipeline_name}__DESTINATION__CREDENTIALS__PRIVATE_KEY"))
    context.log.warning(os.environ.get(f"{config.pipeline_name}__DESTINATION__CREDENTIALS__PROJECT_ID"))

    dlt.secrets[f'{config.pipeline_name}.destination.bigquery.credentials.client_email'] \
        = os.environ.get(f"{config.pipeline_name}__DESTINATION__CREDENTIALS__CLIENT_EMAIL")
    dlt.secrets[f'{config.pipeline_name}.destination.bigquery.credentials.private_key'] \
        = os.environ.get(f"{config.pipeline_name}__DESTINATION__CREDENTIALS__PRIVATE_KEY")
    dlt.secrets[f'{config.pipeline_name}.destination.bigquery.credentials.project_id'] \
        = os.environ.get(f"{config.pipeline_name}__DESTINATION__CREDENTIALS__PROJECT_ID")

    for name, value in os.environ.items():
        context.log.error("{0}: {1}".format(name, value))


    native_value_oauth = {
        "client_id": client_id,
        "client_secret": client_secret,
        "refresh_token": refresh_token,
    }

    json_string_oauth2 = json.dumps(native_value_oauth)
    credentials = GcpOAuthCredentials()
    credentials.parse_native_representation(json_string_oauth2)
    context.log.info(f"GCP OAUTH CREDENTIALS: {credentials.to_native_credentials()}")

    yield from dlt_resource.run(
        dlt_source=google_analytics(
            property_id=config.property_id,
            start_date=config.start_date,
            queries=config.queries,
            credentials=credentials,
        ),
        dlt_pipeline=pipeline(
            pipeline_name=config.pipeline_name,
            destination='bigquery',
            dataset_name=config.dataset_name,
        ),
        context=context,
        dagster_dlt_translator=CustomDagsterDltTranslator()
    )

[mlarose]

You are on the right track. The difference between the dlt_assets and the dlt resources is one produces a declaration of assets that this dlt pipeline would create. And yes it is a multi assets under the hood, where the specifications of the assets are produced by an utility function called build_dlt_asset_specs from the dlt source and pipeline and using the custom translator if you provided one. You can see that here: https://github.com/dagster-io/dagster/blob/master/python_modules/libraries/dagster-embedded-elt/dagster_embedded_elt/dlt/asset_decorator.py#L151.

You can think of the dlt resource as the handle to the tool itself to perform the execution.

Assets declaration serves two purposes:

• declaring what asset exists, what's their specs, dependencies, metadata, etc.
• defining how to materialize them (which is why it uses the dlt resources to run the pipeline in this case)

[mlarose]

If I understand correctly you use a sensor or a schedule to iterate through a list of GA properties in one or several account and you want that to trigger runs of the same asset with different data.

What feels wrong to me is then it's not the same asset. You are thinking of assets as operations, but it should represent a data entity. And if you want them catalogued, then they have to be defined at initialization time.

I am thinking of several directions I can point you.

• use dynamic partitions with your property name or id as a partition_key; the rest of the parameters you can pass as a run config. then you can define your asset op to pass those parameters to dlt_resource.run. But you would have to write your own dlt_source to represent a generalization of the resources that the google_analytics dlt_source normally would so that the Dagster multi-asset can still be properly generated.

• if the list of property does not change that often and could be either a static configuration or something that is generated at initialization time, then you could build an asset factory of dlt_asset. Basically looping through each property and creating one @dlt_assets for each of them.

• if you want to represent the whole thing as a single asset without partition, you could basically roll your own asset that fetches the GA properties, then fans out to fetch each of them with dlt.run.

[bruk0]

@mlarose I want to clarify about third option. Do you mean to remove @dlt_assets decorator and use the @asset instead? And pass parameters to dlt_resource.run(), as I do like this:

yield from dlt_resource.run(
        dlt_source=google_analytics(
            property_id=config.property_id,
            start_date=config.start_date,
            queries=config.queries,
            credentials=credentials,
        ),
        dlt_pipeline=pipeline(
            pipeline_name=config.pipeline_name,
            destination='bigquery',
            dataset_name=config.dataset_name,
        ),
        context=context,
        dagster_dlt_translator=CustomDagsterDltTranslator()
    )

[bruk0]

Mathieu, I appreciate your help and your time you spent on me. I need to go offline, because it's late evening) In case, if you respond on my answer, I won't right back as soon as possible. I wish you a nice Friday and wonderful weekend. Cheers!

[mlarose]

Yes, basically that's what I was suggesting. You can always roll your own implementation if your use case doesn't fit the cookie cutter solutions available. Though, personally first, I'd consider trying with partitions (my first suggestion).

[bruk0]

Hi, @mlarose I've decided to check function build_dlt_asset_specs

build_dlt_asset_specs(
        dlt_source=google_analytics(
                    property_id=config.property_id,
                    start_date=config.start_date,
                    queries=config.queries,
                    credentials=credentials,
                ),
                dlt_pipeline=pipeline(
                    pipeline_name=config.pipeline_name,
                    destination='bigquery', #Check this to load grouped destination secret
                    dataset_name=config.dataset_name,
                ),
        dagster_dlt_translator=CustomDagsterDltTranslator()
    )

It returned me this:

[AssetSpec(key=AssetKey(['dlt_google_analytics_metrics']), deps=[AssetDep(asset_key=AssetKey(['source_secret']), partition_mapping=None), AssetDep(asset_key=AssetKey(['destination_secret']), partition_mapping=None), AssetDep(asset_key=AssetKey(['pipeline_config_asset']), partition_mapping=None)], description='\n Loads data for metrics.\n\n Args:\n metadata: Metadata class object which contains all the information stored in the GA4 metadata.\n\n Yields:\n Generator of dicts, 1 metric at a time.\n ', metadata={'dagster_dlt/source': <dlt.extract.source.DltSource object at 0x131f13a50>, 'dagster_dlt/pipeline': <dlt.pipeline.pipeline.Pipeline object at 0x131f12f90>, 'dagster_dlt/translator': CustomDagsterDltTranslator()}, group_name=None, skippable=False, code_version=None, freshness_policy=None, automation_condition=None, owners=[], tags={'dagster/storage_kind': 'bigquery'}, partitions_def=None), AssetSpec(key=AssetKey(['dlt_google_analytics_dimensions']), deps=[AssetDep(asset_key=AssetKey(['source_secret']), partition_mapping=None), AssetDep(asset_key=AssetKey(['destination_secret']), partition_mapping=None), AssetDep(asset_key=AssetKey(['pipeline_config_asset']), partition_mapping=None)], description='\n Loads data for dimensions.\n\n Args:\n metadata: Metadata class object which contains all the information stored in the GA4 metadata.\n\n Yields:\n Generator of dicts, 1 dimension at a time.\n ', metadata={'dagster_dlt/source': <dlt.extract.source.DltSource object at 0x131f13a50>, 'dagster_dlt/pipeline': <dlt.pipeline.pipeline.Pipeline object at 0x131f12f90>, 'dagster_dlt/translator': CustomDagsterDltTranslator()}, group_name=None, skippable=False, code_version=None, freshness_policy=None, automation_condition=None, owners=[], tags={'dagster/storage_kind': 'bigquery'}, partitions_def=None), AssetSpec(key=AssetKey(['dlt_google_analytics_custom_analytics']), deps=[AssetDep(asset_key=AssetKey(['source_secret']), partition_mapping=None), AssetDep(asset_key=AssetKey(['destination_secret']), partition_mapping=None), AssetDep(asset_key=AssetKey(['pipeline_config_asset']), partition_mapping=None)], description=None, metadata={'dagster_dlt/source': <dlt.extract.source.DltSource object at 0x131f13a50>, 'dagster_dlt/pipeline': <dlt.pipeline.pipeline.Pipeline object at 0x131f12f90>, 'dagster_dlt/translator': CustomDagsterDltTranslator()}, group_name=None, skippable=False, code_version=None, freshness_policy=None, automation_condition=None, owners=[], tags={'dagster/storage_kind': 'bigquery'}, partitions_def=None)]

And I thought that I can build asset factory on top of that response to dynamically generate assets but I didn't succeeded. Is it possible to manage?