op

asset

DagsterInvalidConfigError: Error in config for job

Error 1: Missing required config entry "dummy_asset_partitioned" at path root:ops. Sample config for missing entry: {'dummy_asset_partitioned': {'config': {'date': '...'}}}

GOOGLE_APPLICATION_CREDENTIALS

values.yaml

class PandasCsvIOManagerWithOutputAssetPartitions(IOManager):
    def load_input(self, context):
        file_path = os.path.join("my_base_dir", context.step_key, context.name)
        return pd.read_csv(file_path)

    def handle_output(self, context, obj):
        file_path = os.path.join("my_base_dir", context.step_key, context.name)

        obj.to_csv(file_path, index=False)

        yield <http://MetadataEntry.int|MetadataEntry.int>(obj.shape[0], label="number of rows")
        yield MetadataEntry.float(0.1234, "some_column mean")

    def get_output_asset_key(self, context):
        file_path = os.path.join("my_base_dir", context.step_key, context.name)
        #return AssetKey(file_path)
        return file_path

    def get_output_asset_partitions(self, context):
        return set(context.config["partitions"])

@asset(partitions_def=DailyPartitionsDefinition(start_date="2020-02-01"))
def dummy_asset_partitioned(context) -> DataFrame:
    """Creates a mini dummy asset which is partitioned"""
    partition_key = context.output_asset_partition_key
    get_dagster_logger().info(f"Partitioned asset from: {partition_key}")
    df = pd.DataFrame({'foo':[1,3,3], 'bar':['a', 'b', 'c']})
    df['partition_key'] = partition_key

    rand_metric_dummy_value = random.randrange(0, 101, 2)  
    yield Output(df, metadata={
        "path": EventMetadata.path('/path/to/file'),
        "value_counts": 10,
        "random_dummy_metric": rand_metric_dummy_value
    })

asset_key

dagster.core.errors.DagsterInvariantViolationError: Both the OutputDefinition and the IOManager of output "result" on solid "dummy_asset_partitioned" associate it with an asset. Either remove the asset_key parameter on the OutputDefinition or use an IOManager that does not specify an AssetKey in its get_output_asset_key() function.

yield Output(df, metadata={
        "path": EventMetadata.path('/path/to/file'),
        "value_counts": 10,
        "random_dummy_metric": rand_metric_dummy_value
    })

The op-decorated function accepts DataFrames as parameters and returns DataFrames when it completes. An IOManager handles writing and reading the DataFrames to and from persistent storage.

values_resource

Optional

List

from typing import List
from typing import Optional

from dagster import make_values_resource
from dagster import Enum

from common_definitions import PipelineStage, Pipe  # regular Python enums


# Define a resource for a config that can be passed to multiple ops in a job.
# <https://docs.dagster.io/concepts/configuration/config-schema#passing-configuration-to-multiple-ops-in-a-job>
SHARED_CONFIG = make_values_resource(
    from_stage=Enum.from_python_enum(PipelineStage),  # seems to work
    pipes=Optional[List[Enum.from_python_enum(Pipe)]],  # error
)

/...packages/dagster/core/workspace/context.py:541: UserWarning: Error loading repository location my_pipeline.py:TypeError: Parameters to generic types must be types. Got <dagster.config.config_type.Enum object at 0x7f54f51247c0>.

values_resource

@op(
    out={
        "do_metadata": Out(Nothing, is_required=False),
        "no_metadata": Out(Nothing, is_required=False),
    },
)
def _decide_whether_to_do_metadata(context):
    config = context.resources.shared_config
    if config.should_do_metadata:
        yield Output(None, "do_metadata")
    yield Output(None, "no_metadata")


@op(
    ins={"start": In(Nothing)},
)
def _metadata(context):
    pass  # only do this work if the config said so


@op(
    ins={"start": In(List[Nothing])},
)
def _continue_either_way(context) -> None:
    pass  # always do this work, but wait for metadata to be done


@graph
def _graph():
    do_metadata, no_metadata = _decide_whether_to_do_metadata()
    done_metadata = _metadata(do_metadata)
    _continue_either_way(start=[no_metadata, done_metadata])

_metadata

@op

@op

dagster.check.CheckError: Failure condition: Can not have pending and unresolved step inputs

python_logs:  managed_python_loggers: - root

$DAGSTER_HOME/dagster.yaml

dagit

schedules:
- cron: 0 1 * * *
  extractor: tap-gitlab
  loader: target-snowflake
- cron: 0 2 * * *
  extractor: tap-dbt-cloud
  loader: target-snowflake
- cron: 0 3 * * *
  extractor: tap-salesforce
  loader: target-snowflake
...

context.resources.s3.upload_file_obj(f_obj, "testing", filename)

'S3' object has no attribute 'upload_file_obj'

@op(required_resource_keys={'s3'})

resource_defs={'s3': s3_resource}

upload_file_obj()

@schedule(
    cron_schedule="45 6 * * *",
    job=hello_cereal_job,
    execution_timezone="US/Central",
)

.map(subgraph)

Unexpected dynamic output dependency in regular fan in, should have been caught at definition time.

DagsterInstance

def run_pipeline(...):
    with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml") as temp_config_file:
        shared_config = {
            "a_config_option": x,
            "another_config_option": y,
        }
        yaml.dump(
            {"resources": {"shared_config": {"config": shared_config}}},
            temp_config_file,
        )
        temp_config_file.flush()

        subprocess.check_call([
            "dagster",
            "job",
            "execute",
            "-m", "my.dagster_repo.module",
            "-j", "my_job_name",
            "-c", temp_config_file.name,
        ])

dagster.yaml

AIRBYTE_CONNECTION_ID = "your_airbyte_connection_id"