Can I define my solids in a python class and use them in defining the pipeline using the class object, when I creat my dag?

catherinewu just published a new version: 0.7.14.

There is something that's not entirely clear to me, is there any situation where

dagster-graphql

will run a pipeline directly or does it only communicate that to a

dagit

instance? Also, is it possible to run

dagster

instead of

dagit

to keep a server running or is it only

dagit

that can do it? I find the documentation great, but the architecture isn't completely clear to me by reading it.

Hello Guys, I am new to

DAGSTER

I have just installed it and tried to execute demo code. What I am trying to achieve is as follow, Please excuse me me if I am asking silly question as I am trying to understand it 1. Download Zip file from web URL 2. Unzip it 3. Read CSV from the unzip files 4. Validate CSV (validating data like type,value range etc) and Apply Some Transformation - Join, Merge 5. Save CSV back to specified path Can this work flow possible to implement in Dagster and is there any similar example any one can point me ? Thanks

Is it possible to define global Solid which can be used in any file by just calling it? Something like we just import file and call that solid in DAG file. More real life example If I made

read_csv

solid do I have to copy paste every where in each project where I want to use it or there is any solution to import solids an use them?

@sandy hey, i'm free today to discuss the snowflake dependency issue in https://github.com/dagster-io/dagster/pull/2483 whenever you are

I'm trying out the celery toy example here: https://docs.dagster.io/docs/deploying/celery. For some reason, my celery workers fail to run the job and fail with a

Couldn't import module_name

exception.

I am trying to find some Data Validation Python libraries. I found few but each one have some issue or they are no longer maintained, Any one know better library for CSV and Excel data validation ? https://github.com/roskakori/cutplace https://github.com/TMiguelT/PandasSchema

anyone know how i can fix this,

File "/usr/local/lib/python3.7/site-packages/sqlalchemy/engine/url.py", line 71, in __init__
    self.port = int(port)
ValueError: invalid literal for int() with base 10: ''

seems like the port is empty but this is my config

schedule_storage:
      module: dagster_postgres.schedule_storage
      class: PostgresScheduleStorage
      config:
        postgres_db:
          username:
            env: POSTGRES_USER
          password:
            env: POSTGRES_PASS
          hostname: "<http://dagster-database>"
          db_name:
            env: POSTGRES_DBNAME
          port: 5432

using a local postgres deployment with service, k8s launcher

I’m wondering if there is a way to dynamically add solids to the execution graph? I have a daily schedule that retrieves a set of urls of unknown size and for each URL I would like to retrieve the data from the url, parse the data, and upload the parsed results to our database. While I could do this all in one step it would be nice if each url and each parsing of the results of that url could throw errors and fail without stopping the full process. e.g.

@solid
def get_urls(_):
	url_list = pd.read_csv('urls_to_update.csv')
    # This can be a list of size 0-100
	for url in url_list:
		yield Output(url, "url")

@solid 
def parse_data(_, url):
	res = requests.get(url).json()

	structured_data = dict(
		url=url,
		col_a=res["col_a"],
		col_b=res["col_b"]	
	)

	yield Output(structured_data, 'result')


@solid
def upsert_data(_, data):
	# Some logic to load the data into warehouse 



@pipeline
def my_dynamic_pipeline():
	upsert_data(
		data=parse_data(
			url=get_urls()
		)
	)

Is this just not feasible? Not a huge issue if not I can work around it.

Hi all, I just getting started with dagster and was wondering if someone could answer what is likely a simple question for me: I understand that I can pass an argument to a specific solid via a config file, but how does a solid distinguish which of its parameters it should take from config and which are positional or keyword params from traditional python function syntax?

as a follow up, what if I'd like a configurable argument to be available to all solids in a pipeline? Do I need to specify each solid separately in the config?

does anyone here use python 3.5, or know anyone who does and is unlikely / going to be slow to upgrade? we are trying to gauge the importance of retaining official support & testing on 3.5 vs. the overhead (which is real)

Hello! First of all, congrats for this fantastic tool that is Dagster. I have one little question though: the

dagster_spark

package doesn't seem to have configurations to support Hive (

spark.sql.warehouse.dir

and

.enableHiveSupport

). Will it be possible to do so in an upcoming release ? If not, are you open to PR ?

is there any way to do a branch operator like in airflow? like if no rows returned, dont run rest of solids in pipeline

Hello. I’m trying to get started with Dagster. I’ve worked through the tutorials and have it running in docker. I’m trying to get schedules to work. I have a schedule set and can see them in the UI turned on (and set to run every minute for testing) but the scheduled pipeline don’t seem to be running. I’m sure I’m missing something obvious

Jumping into a k8s-based deployment -- going to drop some initial thoughts into a thread here.

would y'all find it helpful if we had channels for folks interested in kubernetes / ecs / other deployment strategies to chat?

Nate Kupp just published a new version: 0.7.15.

I would like to better understand the the execution model of dagster, RunLauncher, ExecutionPlan, Executor, etc. Where should I start?

Our pipeline uses SGE to distribute jobs on a compute cluster. Can Dagster be used as a layer on top of SGE to submit jobs? Can a Dagster solid run a qsub shell command and somehow understand when the SGE job is finished?

What Python version are you running in production?

Hello 😄 👋 I'm running a pipeline deployed with dask on my local machine and encounter the following error message:

distributed.worker - WARNING - Memory use is high but worker has no data to store to disk.  Perhaps some other process is leaking memory?  Process memory: 907.54 MB -- Worker memory limit: 1.07 GB

As a fix, I have tried to increase the memory limit for my workers to around 3GB for a memory expensive solid:

@dg.solid(
  ...
  tags={'dagster-dask/resource_requirements': {"MEMORY": 3e9}},
)
def parse_dimensions(context, ...):

But, the solid doesn't seem to be executed at all when I run the pipeline. Any insight or general knowledge regarding this type of problem would be very appreciated.

Hi everyone. Just getting started with Dagster but loving it so far. Impressive work! I've been trying to run a small distributed pipeline using multiprocessing, but I'm getting a complain about a missing RunLauncher. This page https://docs.dagster.io/docs/learn/guides/partitioning/backfill states that you indeed need to make sure you have a RunLauncher set up. But I'm not sure how to set one up to run the backfill locally.

are there any plans for having a long running resource like a kafka consumer?

Hi guys. There's a broken link/reference to the repository_partitions decorator (not used any more) on this page https://docs.dagster.io/docs/learn/guides/partitioning/partition_set (in the 0.7.15 docs)

We can register these partitions decorating a function that returns a list of partition set definitions with the Invalid: @repository_partitions

would i run into any issues if i had a pipeline with one long running solid that did stuff and then spun up

execute_pipeline

in another thread or process and loop? like say a kafka consumer spilling to disk after N bytes and then running

execute_pipeline

on a pipeline that finds the file and processes it etc.

Hi, I've been struggling to get the stocks backfill example running:

(dagster) $ DAGSTER_HOME=$PWD dagster pipeline backfill  -y repository.yaml
Select a pipeline to backfill: compute_total_stock_volume: compute_total_stock_volume

     Pipeline: compute_total_stock_volume
Partition set: stock_data_partitions_set
   Partitions: 
                2018-01-01           2018-02-01           2018-03-01           2018-04-01           2018-05-01           2018-06-01           2018-07-01           2018-08-01
                2018-09-01           2018-10-01           2018-11-01           2018-12-01

Do you want to proceed with the backfill (12 partitions)? [y/N]: y
Launching runs... 
Traceback (most recent call last):
  File "/Users/sinayoks/apps/miniconda3/envs/dagster/bin/dagster", line 8, in <module>
    sys.exit(main())
  File "/Users/sinayoks/apps/miniconda3/envs/dagster/lib/python3.6/site-packages/dagster/cli/__init__.py", line 38, in main
    cli(obj={})  # pylint:disable=E1123
  File "/Users/sinayoks/apps/miniconda3/envs/dagster/lib/python3.6/site-packages/click/core.py", line 829, in __call__
    return self.main(*args, **kwargs)
  File "/Users/sinayoks/apps/miniconda3/envs/dagster/lib/python3.6/site-packages/click/core.py", line 782, in main
    rv = self.invoke(ctx)
  File "/Users/sinayoks/apps/miniconda3/envs/dagster/lib/python3.6/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/Users/sinayoks/apps/miniconda3/envs/dagster/lib/python3.6/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/Users/sinayoks/apps/miniconda3/envs/dagster/lib/python3.6/site-packages/click/core.py", line 1066, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/Users/sinayoks/apps/miniconda3/envs/dagster/lib/python3.6/site-packages/click/core.py", line 610, in invoke
    return callback(*args, **kwargs)
  File "/Users/sinayoks/apps/miniconda3/envs/dagster/lib/python3.6/site-packages/dagster/cli/pipeline.py", line 599, in pipeline_backfill_command
    execute_backfill_command(kwargs, click.echo)
  File "/Users/sinayoks/apps/miniconda3/envs/dagster/lib/python3.6/site-packages/dagster/cli/pipeline.py", line 693, in execute_backfill_command
    instance.launch_run(run.run_id)
  File "/Users/sinayoks/apps/miniconda3/envs/dagster/lib/python3.6/site-packages/dagster/core/instance/__init__.py", line 801, in launch_run
    return self._run_launcher.launch_run(self, run)
  File "/Users/sinayoks/apps/miniconda3/envs/dagster/lib/python3.6/site-packages/dagster_graphql/launcher/__init__.py", line 97, in launch_run
    cls=self.__class__.__name__, address=self._address, result=result
dagster.core.errors.DagsterLaunchFailedError: Failed to launch run with RemoteDagitRunLauncher targeting <http://127.0.0.1:3000>:
{'__typename': 'PipelineRunNotFoundError'}

So it looks like the job is creating a run but Dagit isn't able to find it. Any idea what I'm doing wrong here?