dagster #announcements

Ayrton Bourn

12/12/2020, 11:15 AM

As a workaround I thought I could run a separate pipeline for processing each data stream. I was hoping that I'd be able to specify the relevant variables only in the

composite_solid

Copy code

@composite_solid(
    config_schema={
        'source': Field(str, is_optional=True, default_value='BMRS'),
        'stream': Field(str, is_optional=True, default_value='FUELHH'),
        'col_dim': Field(str, is_optional=True, default_value='fueltype'),
    }
)
def update_dataset(context):   
    df_stream_latest = get_latest_df(source, context.solid_config['stream'])
    da_stream_latest = convert_df_to_da(df_stream_latest, context.solid_config['col_dim'])
    save_da_to_zarr(da_stream_latest, source, context.solid_config['stream'])
    
    return 

@pipeline
def update_datasets_pipeline(): 
    update_dataset()
    return

However

composite_solid

can't accept a config_schema, which means I have to repeatedly specify the same variables both in the individual solids, but also the pipeline inputs yaml. E.g.

Copy code

@solid(
    config_schema={
        'source': Field(str, is_optional=True, default_value='BMRS'),
        'stream': Field(str, is_optional=True, default_value='FUELHH')
    }
)
def get_latest_df(context) -> Any:
    df_stream_latest = update.get_stream_latest_data(context.solid_config['source'], context.solid_config['stream'])
    return df_stream_latest

@solid(
    config_schema={
        'col_dim': Field(str, is_optional=True, default_value='fueltype')
    }
)
def convert_df_to_da(context, df_stream_latest: Any) -> Any:
    da_stream_latest = update.convert_df_to_da(df_stream_latest, context.solid_config['col_dim'])
    return da_stream_latest

@solid(
    config_schema={
        'source': Field(str, is_optional=True, default_value='BMRS'),
        'stream': Field(str, is_optional=True, default_value='FUELHH')
    }
)
def save_da_to_zarr(context, da_stream_latest: Any):
    update.save_stream_data_to_zarr(da_stream_latest, context.solid_config['source'], context.solid_config['stream'])
    return

Is there a way to specify these variables only once?

Noah K

12/12/2020, 11:18 AM

I put them in an asset instead 🙂

Noah K

12/12/2020, 11:19 AM

Er sorry, dagster term is resource

Noah K

12/12/2020, 11:19 AM

Copy code

@resource(config_schema={...})
def asset(ctx):
    return ctx.resource_config

Noah K

12/12/2020, 11:19 AM

Then just map that resource into your solids and access it as a dict

Jeremy H

12/14/2020, 3:42 AM

Hello Dagster team, I am using the k8sRunLauncher w/out Celery. It works great when I just have a single node that matches the nodeSelector. However, if I have more than one node, it will also start the dagster jobs on those nodes, but it fails to pull the docker image. Seems that the imagePullSecrets are only getting loaded by the first node. Any ideas? Is this a Dagster issue or am I doing somthing wrong? Error from failed job pod:

Copy code

Events:
  Type     Reason     Age                From               Message
  ----     ------     ----               ----               -------
  Normal   Scheduled  69s                default-scheduler  Successfully assigned dagster/dagster-run-db90fb2b-df8a-4bb2-af55-4cedecb91cb7-nc55c to lke10936-18246-5fd2d5d8e086
  Normal   BackOff    36s (x2 over 67s)  kubelet            Back-off pulling image "<http://docker.pkg.github.com/jeremyhermann/docker-repo/dagster:latest|docker.pkg.github.com/jeremyhermann/docker-repo/dagster:latest>"
  Warning  Failed     36s (x2 over 67s)  kubelet            Error: ImagePullBackOff
  Normal   Pulling    24s (x3 over 68s)  kubelet            Pulling image "<http://docker.pkg.github.com/jeremyhermann/docker-repo/dagster:latest|docker.pkg.github.com/jeremyhermann/docker-repo/dagster:latest>"
  Warning  Failed     24s (x3 over 68s)  kubelet            Failed to pull image "<http://docker.pkg.github.com/jeremyhermann/docker-repo/dagster:latest|docker.pkg.github.com/jeremyhermann/docker-repo/dagster:latest>": rpc error: code = Unknown desc = Error response from daemon: Get <https://docker.pkg.github.com/v2/jeremyhermann/docker-repo/dagster/manifests/latest>: no basic auth credentials
  Warning  Failed     24s (x3 over 68s)  kubelet            Error: ErrImagePull

My Helm values.yaml file:

Copy code

dagit:
  nodeSelector:
    <http://lke.linode.com/pool-id|lke.linode.com/pool-id>: "18034"

postgresql:
  master:
    nodeSelector:
      <http://lke.linode.com/pool-id|lke.linode.com/pool-id>: "18034"

  
k8sRunLauncher:
  enabled: true
  env_secrets:
    - "aws-secrets-env"
  nodeSelector:
    <http://lke.linode.com/pool-id|lke.linode.com/pool-id>: "18386"

userDeployments:
  enabled: true 
  deployments:
    - name: "k8s-user-code"
      image:
        repository: "<http://docker.pkg.github.com/jeremyhermann/docker-repo/dagster|docker.pkg.github.com/jeremyhermann/docker-repo/dagster>"
        tag: latest
        pullPolicy: Always
      dagsterApiGrpcArgs:
        - "-f"
        - "training_repo.py"
      replicaCount: 1  
      port: 3030
      env:
        ENV_VAR: ""
      env_config_maps:
        - ""
      env_secrets:
        - "aws-secrets-env"
      nodeSelector:
        <http://lke.linode.com/pool-id|lke.linode.com/pool-id>: "18386"

      affinity: {}
      tolerations: []
      podSecurityContext: {}
      securityContext: {}
      resources: {}

imagePullSecrets:
  - name: dockerconfigjson-github-com

celery:
  enabled: false

rabbitmq:
  enabled: false

Richard Fisher

12/14/2020, 8:07 AM

Is there a way to configure where the schedule logs are written? By default they seem to be at

$DAGSTER_HOME/schedules/logs/<schedule_id>/

I was hoping to be able to do something like this in the dagster.yaml file:

Copy code

scheduler:
   module: dagster_cron.cron_scheduler
   class: SystemCronScheduler
   config:
      base_dir: /var/lib/persistent_storage

I’ve set up the event_log_storage, which contains most of the same info, but it would be good to have the schedule logs as well.

Istvan Darvas

12/14/2020, 3:23 PM

Hi Guys, I ran into dead ends with hitting the GraphQL API, in a streaming fashion it always freeze after a few hundred hits…. Never mind. So I would like to move / test micro batching / processing, but still using a framework like Dagster (so every 5 minutes would be a scheduled run). But with this approach I will have another problem. How can I run solids parallel in a pipeline where I do not know the inputs only run time. My inputs would be a list of tuple (sensor-id). I think I can put my single sensor pipeline into a composite solid (I have just read a little bit about this) but I still do not understand how should I emit events (device_id_1,device_id_2,…) with Output…lets say I have 5000 sensor id in a micro batch so this should be 5000 parallel composite solid run in a pipeline (If I understand the concept) I can implement a resource coordinator, I think that is not an issue, but I don’t understand how can I achieve this concurrency with Dagster approach if someone can help me my questions would be: - do you have public / example code for something similar ? - the solids will be distributed across the Celery workers? (even those which are part of a composite solid) - Dagit can handle this? (3-5000 composite solids in a pipeline, with 7-8 steps so if I calculate the final task graph, that would be 5000*8 = 40 000 task 😊) Or should I move to a custom solution? Any advice would be appreciated! Thanks, Darvi

sean

12/14/2020, 10:47 PM

Question about dagstermill and intermediates. I have a pipeline that contains various dagstermill solids. The outputs from my solids are xarray DataArrays. When I call

dagstermill.yield_result(my_dataarray, 'name')

at the end of my notebook solid, I get an error-- it appears dagstermill is attempting to serialize the arrays using

scrapbook

, and this fails. How do I bypass scrapbook and use my own serialization logic? What I want to do is just keep an in-memory xarray dataset that stores my arrays. I attempted to do this by creating a custom

asset_store

, but this makes no difference-- dagstermill is still trying to use scrapbook when I call

dagstermill.yield_result

. Here is the code I used for the custom asset store, am I perhaps missing something important?

Copy code

python
import dagster as dg

class DatasetAssetStore(dg.AssetStore):

    def __init__(self):
        super(DatasetAssetStore, self)
        self.dataset = xr.Dataset()

    def get_asset(self, context):
        name = context.output_name
        return self.dataset[name]

    def set_asset(self, context, obj):
        name = context.output_name
        self.dataset[name] = obj

@dg.resource
def dataset_asset_store(_):
    return DatasetAssetStore()

dataset_mode = dg.ModeDefinition(
    resource_defs={"asset_store": dataset_asset_store}
)

my_pipeline = dg.PipelineDefinition(
    name='test',
    solid_defs=[...],
    dependencies={...},
    mode_defs=[dataset_mode]
)

Noah K

12/16/2020, 3:02 AM

Dagster still supports 2.7, right?

Noah K

12/16/2020, 3:03 AM

(not because I am using 2.7, because I need to know if a PR I want to send you needs 2.7 compat)

Nicolas Gaillard

12/16/2020, 10:46 AM

Hello guys, I'm new to Dagster (using 0.9.18 and running inside a container) and I was wondering if it was possible to chain several pipelines ? For example, I want to start my pipeline_2 as soon as pipeline_1 is completed. Thank you in advance and have a nice day!

✅ 1

esztermarton

12/16/2020, 2:51 PM

Hi all! I'm having some trouble with schedules. The schedule seems appropriately configured, and a backfill has successfully run, and partitions all seem generated correctly. However, runs aren't triggered automatically. How/where can I see the logs for whatever is supposed to be triggering so that I can figure out why it isn't triggering? Thanks for the help in advance!

Anatoliy Zhyzhkevych

12/17/2020, 1:53 AM

Hi. Is anyone getting error on analyze_cereal solid in dbt_example running on Windows 10 ( dagster 0.9.22). DBT models run and tests were successful : papermill.exceptions.PapermillExecutionError: --------------------------------------------------------------------------- Exception encountered at "In [2]": --------------------------------------------------------------------------- JSONDecodeError Traceback (most recent call last) <ipython-input-2-37f732a013df> in <module> 9 } 10 ) ---> 11 run_results = __dm_dagstermill._load_parameter('run_results', '{"file": "/tmp/dagstermill/19792f6e-c759-4384-9fb3-0548e89c8ca3/marshal\\input-run_results"}')

Xu Zhang

12/17/2020, 6:32 AM

guys, after i deployed the dagster 0.9.20 to our company’s host over HTTP, the UI can only receive

Succeeded

signal, but for other events, there isn’t any, and also the waterfall wouldn’t stop either. was this due to the HTTP? or do you know what kind of info should i provide to you in order to debug this?

Adonis

12/17/2020, 3:24 PM

Hello, I've been using dagster for a while (great job guys by the way!), however I'm hitting a roadblock: • The backfill using the CLI appears to work properly • The backfill using dagit doesn't do anything (no log, no action after clicking the button) Has anyone here already experienced this issue? Does anyone can tell me how to have debug/verbose mode enabled when using dagit? Many thanks! Tested on 0.9.21 and 0.9.22.post0 Previously the backfill with the UI on 0.9.5 worked properly

Frank Dekervel

12/17/2020, 4:16 PM

Hello, question, does dagster already support skipping a solid if its output already exists ? a bit like dvc / luigi / makefiles

Oleg Agapov

12/17/2020, 5:42 PM

Hi guys, One thing I don't understand about Dagster. What should I do if I need to make dependent solids if they don't have outputs? Similar in Airflow is

Copy code

task1 >> task2 >> task3

My solids are kinda stateless (don't return any values), but their order is important.

Jacolon Walker

12/19/2020, 8:05 AM

Hey all! Quick question with workspaces.yaml. I am trying to load an external python environment and for some reason I am getting a weird error.

Copy code

Error 1: You can only specify a single field at path root:load_from[0]. You specified ['executable_path', 'python_environment', 'target']. The available fields are ['grpc_server', 'python_environment', 'python_file', 'python_module', 'python_package']

Jacolon Walker

12/19/2020, 8:06 AM

for reference I only have load_from, python_environment, executable_path and target

Noah K

12/19/2020, 8:28 AM

What does your workspace look like?

Jacolon Walker

12/19/2020, 8:29 AM

Ah looking at it now looks like its a spacing issue

Jacolon Walker

12/19/2020, 8:32 AM

yup that is what it was

Jacolon Walker

12/19/2020, 8:32 AM

🙂

Daniel Kim

12/19/2020, 2:53 PM

Hello, I saw this great

video▾

where Nick talks about displaying structured event logs in the dagit UI, but I can't seem to find the docs on how to implement this or is this a relatively new feature that is not yet implemented? Can it be used with the default sqlite backend?

Rolf Schick

12/20/2020, 1:12 PM

Hi 👋 Has anyone got the k8s user-code example from @Michiel Ghyselinck recently working? https://dev.to/michielhub/dagster-with-user-code-deployments-grpc-2c16

sk4la

12/20/2020, 7:24 PM

Hey everyone! 👋 I’ve got a quick question about logging inside Dagster. Coming from Prefect, I’m used to “externalizing” some portions of code as foreign functions (understand “outside of the Prefect scope”) in order to reuse functionalities from within several tasks, which helps minimizing the overall code redundency. Basically, I’m just calling generic functions from within several tasks, like filesystem-related operations, etc. In order to log stuff from within these foreign functions, Prefect exposes the main logger as a contextual object, so you can use it outside of a Prefect-declared task like this:

Copy code

# Prefect example
import logging
import prefect

@task
def call_a_foreign_function():
    log_from_a_foreign_function()

def log_from_a_foreign_function():
    logger = prefect.context.get("logger", logging.getLogger("default_logger_if_no_prefect_context_is_found"))
    <http://logger.info|logger.info>("It works! I can still log to the main logger even if I'm a foreign function!")

I tried to do something like this in order to do the equivalent within Dagster:

Copy code

# Dagster equivalent
import logging
from dagster import solid

@solid
def call_a_foreign_function():
    log_from_a_foreign_function()

def log_from_a_foreign_function():
    logging.getLoggerClass()("dagster").warning("Can I speak to the UI like this?")
    logging.getLogger("dagster").warning("Can I speak to the UI like this?")

Sadly, it does not seem to work, I’m only able to log to the Dagit console (not the UI). Does Dagster have a cool way to log to the UI from “outside” of solids ?

Rolf Schick

12/21/2020, 12:25 PM

Hi I try to use GCS for intermediate storage in the K8s Celery example. It seams the https://docs.dagster.io/deploying/gcp is not working on the K8s Helm deployment. (It works locally) Appreciate any help / hint.

esztermarton

12/21/2020, 2:27 PM

Hi everyone! Is there a good way to, from within a solid, know where we are in the pipeline? I.e. to know what solids have already happened and what solids remain downstream?

schrockn

12/21/2020, 5:07 PM

Latest post from @dwall! This one is a doozy: https://twitter.com/dagsterio/status/1341065723205054465