19913d85-f662-4518-a9a1-7adc1793dfd4.png){kind=small}
Channels
dagster-de
dagster-cube
gigs-freelance
dagster-wandb
dagster-machine-learning
dagster-bigquery
tools
dagster-dask
dagster-snowflake
dagster-releases
data-platform-design
events
dagster-serverless
dagster-noteable
introductions
dagster-airflow
dagster-cloud
random
dagster-dbt
faq-read-me-before-posting
dagster-airbyte
dagster-spatial
dagster-support
announcements
github-discussions
豆瓣酱帮
dagster-feedback
dagstereo
dagster-ecs
dagster-kubernetes
jobs
dagster-showcase
Title
s
Stanley Yang
10/18/2022, 7:43 PMis there a list of things to avoid when doing multi-threaded work within an op?
The context is that we are using dask's ThreadPoolExecutor to spawn 4 threads. We noticed logs were being dropped, which led me to this related thread where I learned the context object not being thread-safe.
A related question is, if we want to have threadsafe logging, should we use the default logging module instead of the one provided by context?
a
alex
10/20/2022, 5:09 PMhmm, while some aspects of the context may have thread safety issues im surprised it was dropping logs since the underlying python logging should be thread safe. May be an issue with our handler i suppose. Do you have any more detail about the dropped logs?
Using the default loggers should be an option, stdout/stderr get captured (which we refer to as “compute logs”) so the contents should still be available in dagit, just not in the main structured event stream.
Generally you should be safe to use threads as you please. Generally i would expect them to be
joinr
Rohan Prasad
10/20/2022, 9:54 PMYea, what we were doing here is trying to get some logs from a thread back into the dagit main structured event stream.
The rough code was using a ThreadPoolExecutor and passing the context object to the function called in the thread (where it called
<http://context.log.info|context.log.info>()def func(l, context):
<http://context.log.info|context.log.info>("Log something")
# do something
r = []
with ThreadPoolExecutor(thread_count):
for l in ll:
r.append(func(l, context))that being said we could use the standard python logger here and viewing the results in the “compute logs” - viewing them in the dagit main structured stream is more of a nice to have
a
alex
10/20/2022, 10:15 PMyou could have the main thread poll a
Queuestill not obvious to me why the messages were dropped without more details
r
Rohan Prasad
10/21/2022, 2:11 PMyea, I’m not really sure why either - let me keep a closer tab on it the next time I run this flow and if I see anything else that may help - I’ll share it here