Dagit in Docker

[galloramiro]

Hi my name is Ramiro and Im tring to make Dagster and Dagit run in a docker and docker compose environment.

Im following the Runnin Dagit as a service Documentation

But the process fails and freeze.
When I run this in a pipenv with the command dagit -f src/tasks/schedule_definition.py runs flowerless

But when I put this on a docker it fails
Here I share you my code
DockerFile

FROM python:3.8-slim

RUN apt-get update

RUN apt-get install -y \
        build-essential \
        libgflags-dev \
        libsnappy-dev \
        zlib1g-dev \
        libbz2-dev \
        liblz4-dev \
        librocksdb-dev \
        libzstd-dev \
        cron

# Create the workdir
WORKDIR scrape_web_tasks

CMD mkdir scrape_web_tasks/src
COPY requirements.txt src/
RUN pip install -r src/requirements.txt

ADD src/ src/

ENV SRCDIR "/scrape_web_tasks/src/"
ENV DAGSTER_HOME "${SRCDIR}"
ENV PYTHONPATH "${SRCDIR}"

RUN chmod +x src/entrypoint.sh

EXPOSE 3000

ENTRYPOINT ["src/entrypoint.sh"]

entrypoint.sh

#!/bin/sh
# This block may be omitted if not packaging a repository with cron schedules:
####################################################################################################
# see: https://unix.stackexchange.com/a/453053 - fixes inflated hard link count
touch /etc/crontab /etc/cron.*/*

service cron start

# Add all schedules defined by the user
dagster schedule up
####################################################################################################

# Launch Dagit as a service
dagit -f src/tasks/schedule_definition.py -h 0.0.0.0 -p 3000

dagster.yaml

scheduler:
  module: dagster_cron.cron_scheduler
  class: SystemCronScheduler

pipelines.py

from datetime import date

from dagster import solid, Failure, pipeline
from dateutil.relativedelta import relativedelta

from scrapper.scrappers.factories import ScrappersFactory
from scrapper.services.repositories import MongoDbRepository
from tasks.prefect_tasks import mercado_libre_scrapper, easy_scrapper, afcp_scrapper


@solid
def mercado_libre_scrapper(context):
    scrapper_factory = ScrappersFactory()

    scrapper = scrapper_factory.build_mercado_libre_scrapper()
    scrapper_succeed = scrapper.run()

    if scrapper_succeed[0]:
        context.log.info(scrapper_succeed[1])
    else:
        context.log.info(scrapper_succeed[1])
        raise Failure(scrapper_succeed[1])


@solid
def easy_scrapper(context):
    scrapper_factory = ScrappersFactory()

    scrapper = scrapper_factory.build_easy_scrapper()
    scrapper_succeed = scrapper.run()

    if scrapper_succeed[0]:
        context.log.info(scrapper_succeed[1])
    else:
        raise Failure(scrapper_succeed[1])


@solid
def afcp_scrapper(context, day_limit: int = 15):
    repository = MongoDbRepository()

    today = date.today()
    date_to_scrapp = today - relativedelta(months=1)

    year = date_to_scrapp.strftime('%Y')
    month = date_to_scrapp.strftime('%m')
    data_saved = repository.afcp_monthly_data_was_saved(year=year, month=month)

    if data_saved:
        context.log.info('The values was saved before')
        return True

    if date_to_scrapp.day > day_limit and not data_saved:
        context.log.info('The first 15 days of the month have pass and we dont get the values')
        raise Failure('The first 15 days of the month have pass and we dont get the values')

    if date_to_scrapp.day > day_limit and data_saved:
        context.log.info('The first 15 days of the month have pass, and we get the values')
        return True

    scrapper_factory = ScrappersFactory()
    scrapper = scrapper_factory.build_afcp_scrapper()

    scrapper_succeed = scrapper.run(year=year, month=month)

    if scrapper_succeed[0]:
        context.log.info(scrapper_succeed[1])
    else:
        raise Failure(scrapper_succeed[1])


@pipeline
def cement_scrapper_pipeline():
    mercado_libre_scrapper()
    easy_scrapper()
    afcp_scrapper()

schedule_definition.py

from datetime import time, datetime

from dagster import daily_schedule, repository

from tasks.pipelines import cement_scrapper_pipeline


@daily_schedule(pipeline_name="cement_scrapper_pipeline", start_date=datetime(2021, 1, 1), execution_time=time(hour=7))
def daily_cement_scrapper_schedule(date):
    return dict(solids=dict(
        mercado_libre_scrapper=dict(date=dict(value=date.strftime("%Y-%m-%d"))),
        easy_scrapper=dict(date=dict(value=date.strftime("%Y-%m-%d"))),
        afcp_scrapper=dict(date=dict(value=date.strftime("%Y-%m-%d"))),
        ),
    )


@repository
def cement_scraper_repository():
    return [cement_scrapper_pipeline, daily_cement_scrapper_schedule]

And here is the output:

docker run 'lomak8s/scrape_web_tasks' dagit -f /src/tasks/schedule_definition.py -h 0.0.0.0 -p 3000
Starting periodic command scheduler: cron.
Traceback (most recent call last):
  File "/usr/local/bin/dagster", line 8, in <module>
    sys.exit(main())
  File "/usr/local/lib/python3.8/site-packages/dagster/cli/__init__.py", line 41, in main
    cli(auto_envvar_prefix=ENV_PREFIX)  # pylint:disable=E1123
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 829, in __call__
    return self.main(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 782, in main
    rv = self.invoke(ctx)
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1259, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 1066, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/usr/local/lib/python3.8/site-packages/click/core.py", line 610, in invoke
    return callback(*args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/dagster/cli/schedule.py", line 173, in schedule_up_command
    return execute_up_command(preview, kwargs, click.echo)
  File "/usr/local/lib/python3.8/site-packages/dagster/cli/schedule.py", line 178, in execute_up_command
    with get_external_repository_from_kwargs(cli_args) as external_repo:
  File "/usr/local/lib/python3.8/contextlib.py", line 113, in __enter__
    return next(self.gen)
  File "/usr/local/lib/python3.8/site-packages/dagster/cli/workspace/cli_target.py", line 666, in get_external_repository_from_kwargs
    with get_repository_location_from_kwargs(kwargs) as repo_location:
  File "/usr/local/lib/python3.8/contextlib.py", line 113, in __enter__
    return next(self.gen)
  File "/usr/local/lib/python3.8/site-packages/dagster/cli/workspace/cli_target.py", line 593, in get_repository_location_from_kwargs
    with RepositoryLocationHandle.create_from_repository_location_origin(origin) as handle:
  File "/usr/local/lib/python3.8/site-packages/dagster/core/host_representation/handle.py", line 47, in create_from_repository_location_origin
    return ManagedGrpcPythonEnvRepositoryLocationHandle(repo_location_origin)
  File "/usr/local/lib/python3.8/site-packages/dagster/core/host_representation/handle.py", line 188, in __init__
    list_repositories_response = sync_list_repositories_grpc(self.client)
  File "/usr/local/lib/python3.8/site-packages/dagster/api/list_repositories.py", line 16, in sync_list_repositories_grpc
    raise DagsterUserCodeProcessError(
dagster.core.errors.DagsterUserCodeProcessError: (ModuleNotFoundError) - ModuleNotFoundError: No module named 'tasks'

Stack Trace: 
  File "/usr/local/lib/python3.8/site-packages/dagster/grpc/server.py", line 343, in ListRepositories
    self._repository_symbols_and_code_pointers.loadable_repository_symbols,
  File "/usr/local/lib/python3.8/site-packages/dagster/grpc/server.py", line 105, in loadable_repository_symbols
    self.load()
  File "/usr/local/lib/python3.8/site-packages/dagster/grpc/server.py", line 95, in load
    self._loadable_repository_symbols = load_loadable_repository_symbols(
  File "/usr/local/lib/python3.8/site-packages/dagster/grpc/server.py", line 119, in load_loadable_repository_symbols
    loadable_targets = get_loadable_targets(
  File "/usr/local/lib/python3.8/site-packages/dagster/grpc/utils.py", line 25, in get_loadable_targets
    else loadable_targets_from_python_file(python_file, working_directory)
  File "/usr/local/lib/python3.8/site-packages/dagster/cli/workspace/autodiscovery.py", line 11, in loadable_targets_from_python_file
    loaded_module = load_python_file(python_file, working_directory)
  File "/usr/local/lib/python3.8/site-packages/dagster/core/code_pointer.py", line 190, in load_python_file
    raise error
  File "/usr/local/lib/python3.8/site-packages/dagster/core/code_pointer.py", line 138, in load_python_file
    module = import_module_from_path(module_name, python_file)
  File "/usr/local/lib/python3.8/site-packages/dagster/seven/__init__.py", line 118, in import_module_from_path
    spec.loader.exec_module(module)
  File "<frozen importlib._bootstrap_external>", line 783, in exec_module
  File "<frozen importlib._bootstrap>", line 219, in _call_with_frames_removed
  File "/src/tasks/schedule_definition.py", line 5, in <module>
    from tasks.pipelines import cement_scrapper_pipeline

Freeze loading the pipeline... I dont understunt

If anyone can help I will thank you a lot!

[gibsondan]

Hi Ramiro - it looks like it's having some trouble finding the 'tasks' module. There's a -d argument that you can use to specify a specific working directory that it should use - from the example it seems like that might be '/src/'?

So I'd be curious if

dagit -f /src/tasks/schedule_definition.py -d /src/ -h 0.0.0.0 -p 3000

would work any better.

[galloramiro]

Thanks gibsondan for the answer!

Try with -d but the problem wasnt solved.

I solve the importation by editing the DockerFile, the update version is in the original post, and update the entry point.
If I add the -d on the entry point the importation problems apears again.
But now this problem apears:

Starting periodic command scheduler: cron.
Usage: dagster schedule up [OPTIONS]

Error: No arguments given and workspace.yaml not found.

I have a workspace.yaml in the same level that the dagster.yaml but its seams to not recognice it, I seek in the documentation but they say that will be auto searched (english is not my primary langauge, srry)

workspace.yaml

load_from:
  - python_file: tasks/schedule_definition.py

If coment the entrypoint on the DockerFile and the run manualy the command:

$ docker run 'lomak8s/scrape_web_tasks' dagit -f src/tasks/schedule_definition.py -h 0.0.0.0 -p 3000

It freez... and when I stop the container I have the following response:

^CKeyboardInterrupt
2021-01-13T15:02:22Z

Aborted!
Loading repository...
Serving on http://0.0.0.0:3000 in process 1