►
From YouTube: Manage you data pipelines with Dagster | Software defined assets | IO Managers | Updated project
Description
In this video we will revisit dagster. We will talk about changes to this workflow orchestration system due to recent updates (update from version 0.15 to 1.3.1)
Dagster is an orchestrator that's designed for developing and maintaining data assets, such as tables, data sets, machine learning models, and reports.
We will cover Software Defined Assets as dagster is pushing towards the Software Defined Assets. By default our pipeline outputs are stored as a pickle file in the dagster home folder. What if we want to store the outputs in a database table, or in a readable file such as a csv or parquet file. Dagster provides us with Input and Output managers (IO managers) that enable reading and writing data to storage systems. Using Store IO managers we can save the outputs in a file system or store our data as tables in a database. We will define file csv/parquet and database IO Managers.
Link to previous video: https://www.youtube.com/watch?v=t8QADtYdWEI&t
Link to GitHub repo: https://github.com/hnawaz007/pythondataanalysis/tree/main/dagster-project/etl
Get started with Dagster in just three quick steps:
Install Dagster, Define assets and Materialize assets.
Create a virtual environment: python -m venv env
Activate the virtual environment: env\Scripts\activate
To install Dagster into an existing Python environment, run:
pip install dagster dagit
Command to create a new project
dagster project scaffold --name my-dagster-project
Additional libraries required: Pandas, psycopg2
To run dagster issue following command:
dagit
dagster-daemon run
Access Dagit UI on port 3000: http://127.0.0.1:3000
💥Subscribe to our channel:
https://www.youtube.com/c/HaqNawaz
📌 Links
-----------------------------------------
#️⃣ Follow me on social media! #️⃣
🔗 GitHub: https://github.com/hnawaz007
📸 Instagram: https://www.instagram.com/bi_insights_inc
📝 LinkedIn: https://www.linkedin.com/in/haq-nawaz/
🔗 https://medium.com/@hnawaz100
-----------------------------------------
#Python #ETL #Dagster
Topics covered in this video:
==================================
0:00 - Introduction to Dagster
2:11 - Dagster create new project
3:03 - Dagster Project Structure
4:18 - Software Defined Assets
5:35 - Install Required Libraries
5:58 - Source DB Connection
6:27 - Source Asset
10:05 - File IO Manager
14:16 - Second Asset
16:19 - Parquet IO Manager
16:26 - Database IO Manager
19:05 - Materialize Assets
A
Hello
and
welcome
one
and
all
in
this
session
we
will
revisit
Daxter.
We
will
talk
about
changes
to
this
workflow
orchestration
system
due
to
recent
updates.
Daxter
is
an
orchestrator
best
designed
for
developing
and
maintaining
data
assets
such
as
tables
data
sets
machine
learning
models
and
reports.
In
the
previous
video,
we
created
a
Dexter
project
using
Daxter
version
0.15
in
that
project.
We
converted
our
python
ETL
data
pipeline
to
adagster
compatible
code.
A
In
doing
so,
we
covered
main
Dexter
Concepts
such
as
op
jobs
and
schedule,
and
we
brought
the
various
components
together
in
a
repository
module.
Much
has
changed
since
then,
so
we
will
create
a
new
project
and
review
its
structure.
We
will
cover
software-defined
assets
as
dagster
is
pushing
towards
software-defined
assets.
A
Finally,
we
will
store
the
outputs
of
the
assets
or
the
data
pipeline
into
a
permanent
storage
for
later
use
by
default.
Our
pipeline
outputs
are
stored
as
pickle
file
in
the
diagster
home
folder.
What
if
we
want
to
store
the
outputs
in
a
database
table
or
in
a
readable
file
format
such
as
CSV
or
parquet?
Previously
we
pushed
the
data
to
a
database
via
custom
python
script.
A
A
They
manage
the
input
by
reading
in
assets
from
where
the
data
is
stored
and
loaded
into
memory
to
be
used
by
a
dependent
asset.
We
will
Define
a
file
and
a
database.
I
o
manager,
let's
check
our
environment
and
see
the
available
Target
and
dagster
versions.
At
the
time
of
the
recording.
We
are
working
with
version
1.3.1.
A
A
However,
in
the
new
version
1.3.1
the
command
to
create
a
new
project
is
as
follow
Dexter
project
and
we
have
a
keyword,
scaffold
dash,
dash
name,
then
the
project
name,
so
we
will
provide
Dexter
project
scaffold,
dash,
dash,
name
ETL.
Let's
review
the
project
structure
in
this
version,
as
you
can
see,
there
is
no
workspace.gml
file.
It
is
replaced
by
a
pi
project.tamil
file.
A
A
As
our
project
grows,
it
is
recommended
to
organize
Assets
in
sub
packages
or
sub
modules.
Just
like
in
the
previous
version.
We
can
create
folders
and
move
our
code
inside
those
folders
in
the
assets
file.
We
Define
our
ETL
logic.
This
is
a
very
minimalistic
setup.
Just
the
base
required
code
to
get
you
up
and
running.
A
We
will
develop
an
ATL
Pipeline
with
dagster
software
defined
assets.
We
will
ingest
data
from
external
sources,
explore
and
transform
that
data
and
then
materialize
the
output
that
we
can
consume
later.
Let's
start
by
defining
what
are
software-defined
assets.
A
software-defined
asset
is
an
object
in
persistent
storage.
A
Software-Defined
asset
is
a
declarative
approach
to
data
management,
meaning
we
write
the
code
that
a
product
table
should
exist
in
adventureworks
database
and
how
this
table
should
be
computed.
A
software-defined
asset
can
be
referenced
by
an
asset.
Key
and
asset
key
is
a
handle
for
referring
to
the
asset,
such
as
a
file
or
a
table.
A
In
addition,
an
asset
has
a
set
of
Upstream
asset
key.
This
is
a
pointer
to
the
previous
asset
that
helps
derive
the
current
asset.
For
example,
if
you
have
a
source
table,
the
current
Table
will
have
a
pointer
to
the
source
table,
so
it
can
read
data
from
the
source
and
then
build
the
current
Table.
A
We
will
see
an
example
of
this
in
our
project
now
that
we
have
a
project
created
and
we
know
what
assets
are
it's
time
for
us
to
build
the
first
part
of
our
data
pipeline
before
jumping
into
the
code.
Let's
install
the
required
libraries
in
our
virtual
environment.
We
have
the
libraries
listed
in
a
requirements
file.
We
will
execute
a
pip
command
with
Dash
R
and
provided
our
requirements
file.
This
will
go
ahead
and
install
the
required
libraries.
We
will
borrow
much
of
the
code
from
the
previous
session.
A
We
still
need
a
database
connection
to
extract
data
from
the
source.
In
the
previous
session,
we
defined
a
DB
underscore
conf
file,
we'll
copy
this
file
in
the
current
project,
and
we
will
save
this
in
our
resources
folder
in
this
file,
we
only
keep
the
SQL
Server
Connection.
This
is
our
source.
We
will
handle
our
Target
postgres
via
I
o
manager,
let's
open
the
assets
file
and
start
defining
our
assets.
A
We
can
change
the
name
of
this
file
to
something
appropriate.
For
example,
we
can
give
it
the
same
name
as
our
table
as
we
build
more
assets.
This
will
help
us
distinguish
between
them.
Things
can
get
quite
complex.
If
you
have
hundreds
of
files,
we
will
rename
this
to
assets
underscore
product
category.
In
this
file.
A
We
begin
with
asset
decorator.
Under
this
we
Define
a
function
that
will
read
data
from
the
source
we
provide
a
name
for
this
function,
extract
dim
product
category.
This
function
takes
context
as
an
argument.
This
is
a
special
object
and
it
exposes
the
extra
system,
API,
resources,
access
and
logging.
A
We
can
use
contacts
as
the
first
argument
of
a
compute
function.
This
is
followed
by
the
type
of
return
value
from
this
function.
It
will
be
a
data
frame,
meaning
this
function
will
return
a
data
frame.
We
call
the
get
SQL
con
function
in
a
width
statement.
This
gives
us
access
to
the
SQL
Server
Connection.
A
We
utilize
the
read,
SQL
query
function
from
pandas
and
pass
it
a
SQL
statement
along
with
our
connection,
this
returns
a
data
frame
that
we
save
into
a
DF
variable.
Finally,
we
return
this
variable.
This
completes
the
extract
function
in
our
virtual
environment.
Let's
go
ahead
and
launch
dagster
UI
with
Target
command
in
the
UI.
We
can
interact
with
our
asset.
A
We
will
execute
the
asset
and
see
it
materialized.
We
will
see
the
default
I
o
manager
in
action,
and
this
I
o
manager
writes
this
file
to
our
dagster
home
location.
The
Dexter
home
is
set
via
environment
variable.
We
can
copy
the
location
of
the
file
and
view
it
by
default.
The
outputs
are
stored
in
a
pickle
file.
A
We
can
read
the
pickle
file
with
pandas
if
we
want
to
review
the
output
anyways,
let's
make
some
update
to
our
asset,
let's
add
some
tags
to
it.
So
back
in
vs
code,
we
assign
a
group
name
to
this
asset.
This
way
we
can
group
The
related
assets
together.
In
addition,
we
can
add
a
compute
kind
of
pandas.
A
This
will
display
the
return
type
of
this
asset
in
the
dagget
UI.
We
also
log
the
first
five
rows
of
the
data
frame,
so
we
get
a
preview
of
the
data
with
these
changes
in
place,
let's
save
our
file
and
reload
our
asset
definition
in
the
UI.
We
see
our
group
name
in
the
left,
Pane
and
below
the
asset.
We
see
a
pandas
tag.
This
tells
us
that
this
asset
returns
a
panda's
object.
We
materialize
the
asset
and
this
time
around,
we
get
a
preview
of
the
data
frame.
A
Okay,
we
have
our
extract
part
of
the
ETL
pipeline.
Let's
go
ahead
and
Define
our
first
custom.
I
o
manager
to
change
the
persistent
storage
for
this
asset.
We
will
change
the
output
of
this
asset
from
a
pickle
file
to
a
CSV
file
with
a
custom
I
o
manager.
So
we
can
open
the
file
in
Excel
and
review
it.
A
A
In
this
folder,
we
will
create
a
new
python
file.
I
will
call
it
file
underscore
IO
in
this
file.
We
import
the
required
libraries.
We
start
by
defining
a
class
called
local
file
system.
I
o
manager-
and
this
takes
I
o
manager
as
an
argument.
Under
this
class
we
Define
three
function
in
the
first
function.
We
Define
the
path
of
the
file.
A
A
If
you
recall,
an
asset
key
is
a
handle
for
referring
to
the
asset
and
the
return
type
of
this
function
will
be
a
string.
We
declare
a
variable
our
path
with
the
help
of
join
function,
OS
dot
path.
We
combine
a
folder
location
with
asset
key
and
append
the
extension
which
is
CSV.
So
this
is
the
makeup
of
our
file
path.
Let's
create
the
warehouse
underscore
location
and
the
subfolder
inside
it
in
the
root
directory
of
our
project.
A
It
will
be
under
the
first
ETL
folder.
We
create
a
warehouse,
underscore
location,
folder
and
inside
it
a
result.
Folder,
the
output
of
our
asset
will
be
stored
here
back
in
our
code.
We
return
the
absolute
path
of
the
current
working
directory
with
file
name.
The
second
function
saves
the
output
of
the
asset
in
this
function.
We
pass
the
context
and
the
data
frame
along
with
self.
A
We
call
the
first
function
to
obtain
the
file
path,
Vlog
info
for
the
UI
and
finally
save
the
data
frame
with
two
underscore
CSV
function.
In
the
last
function.
We
get
the
file
path
and
read
the
file
into
a
data
frame
and
return
this
data
frame.
Finally,
we
defined
a
function
that
returns
this
class.
A
So
this
is
our
custom.
I
o
manager
that
handles
the
output
of
our
asset.
Let's
implement
this
in
our
code
in
the
asset
we
Define
an
I
o
manager
key
and
set
this
to
a
value
of
file
underscore
I
o
in
the
init
file
we
import
from.
I
o
file,
I
o
module
in
the
definition
we
Define
resources
and
set
the
key
value
for
I
o
manager.
A
This
must
be
the
same
value
we
provided
in
the
asset
we
set
this
to
file
underscore
I
o
and
call
our
class
from
this
module
our
custom
I
o
manager
is
implemented.
Let's
save
all
the
files
we
have
touched
and
head
to
Dag
its
UI,
let's
reload
our
asset
definition,
there's
no
apparent
change
in
the
UI.
A
Let's
go
ahead
and
materialize
our
asset
and
see
if
the
asset
output
is
saved
to
the
new
location
as
a
CSV
file
in
the
logs,
we
can
see
that
asset
is
utilizing.
The
file
underscore
I
o
module,
which
is
a
good
sign.
We
have
implemented
the
I
o
manager
correctly
and
we
see
the
new
file
path
and
it
is
pointing
to
the
correct
location,
meaning
our
asset
output
is
saved
as
a
CSV
file.
We
can
open
the
file
location,
open
the
file
and
view
its
content
to
make
sure
it
has
valid
data.
A
Let's
define
our
second
asset,
similar
construct,
we
Define
a
function
and
to
this
function
we
provide
the
name
of
the
function
from
the
first
asset.
This
will
Define
the
dependency
between
the
two
assets.
This
function
will
also
return
a
data
frame
we
perform
few
Transformations,
such
as
select
required,
columns
and
rename
a
column.
We
will
go
ahead
and
return
this
revised
data
frame.
Let's
wrap
this
function
up
with
except
block.
A
Let's
save
this
and
view
it
in
the
UI.
We
see
our
second
asset
and
there's
a
dependency
between
the
two.
We
can
materialize.
The
second
asset
to
confirm
our
Transformations
are
correct.
This
is
one
plus
point
of
dixter.
We
don't
have
to
run
all
the
steps
if
all
the
assets
are
materialized
and
one
of
them
fails
or
have
a
code
update.
Then
we
can
fix
the
error
in
that
asset
and
rerun
from
that
step
forward.
A
If
our
first
asset
takes
a
long
time
to
run,
then
we
can
only
materialize
it
once
and
rerun
subsequent
assets
multiple
times.
Let's
select
and
materialize
our
second
asset,
it
runs
successfully,
and
we
can
view
the
file
to
confirm
our
transformation
steps.
We
have
deleted
a
few
columns
in
this
step
and
renamed
a
column.
All
looks
good.
If
you're
working
with
a
larger
data
set,
then
you
can
save
your
files
to
parquet
format
rather
than
CSV.
A
A
We
materialize
the
assets
and
see
the
outcome.
In
this
run
instance,
we
see
that
our
assets
are
materialized
as
parquet
files.
Let's
go
ahead
and
Define
a
second
I
O
manager
that
will
save
our
asset
outputs
to
a
database.
This
will
follow
the
same
construct
as
our
file.
I
o
manager,
rather
than
writing
it
from
scratch.
I'll
copy
paste
the
pre-written
code
and
we
will
go
over
it.
We
have
a
class
with
three
functions
in
the
first
function.
We
get
a
hold
of
the
credentials
in
the
second
function.
A
We
handle
the
asset
output,
we
Define
the
table
name
with
the
asset
key
and
save
the
data
frame
to
a
database
table
with
two
SQL
function.
We
will
skip
over
the
SQL
Alchemy
as
we
have
covered
this
number
of
times
on
this
channel
in
the
third
function.
We
get
a
hold
of
the
Upstream
output
and
retrieve
the
asset
key,
so
we
want
to
make
sure
we
pass
the
correct
table
name.
We
query
the
database
table
and
return
it
as
a
data
frame.
Here
we
Define
a
decorator
of
type,
I
o
manager.
A
A
We
don't
want
to
hard
code
the
credentials,
especially
if
we
check
this
project
in
a
git
repo
also
make
sure
to
add
the
EnV
file
to
git,
ignore
let's
quickly
Define
an
EnV
file
and
add
our
environment
variables
to
it.
This
file
does
not
have
an
extension.
So
make
sure
to
delete
the
extension
if
you're
creating
a
new
file.
A
So
this
is
the
content
of
EnV
file
and
once
we
reload
dag
it
you'll
see
the
environment.
Variable
defined
here,
they'll
be
loaded
to
the
dagget
UI.
Let's
go
ahead
and
save
the
file,
we
will
stop
and
restart
the
decade
UI.
To
pick
up
this
new
file
on
the
restart,
you
see
that
it
has
loaded
the
environment
variables
from
this
file.
Now
we
can
Implement
our
database.
I
o
manager
in
the
asset
change
the
I
o
manager.
Key
I
will
only
change
it
for
the
second
asset.
A
Now
in
the
init
file,
we
need
to
First
import
this.
I
o
manager
and
then
configure
it.
We
provide
the
key
for
this.
I
o
manager
and
then
make
a
call
to
dbio
manager
and
invoke
the
function
since
the
function
requires
a
dictionary
with
credentials.
We
pass
it
a
dictionary
with
credentials
from
the
environment
variables
our
database,
I
o
manager,
is
implemented,
let's
save
it
and
give
it
a
run
as
usual
reload
all
of
our
asset
definition
to
make
sure
we
have
the
updated
code,
let's
go
ahead
and
materialize.
A
A
Okay,
we
see
a
reference
to
adventureworks
database
and
a
table
name,
so
this,
indeed,
is
using
our
database
underscore
I
o
module.
At
this
point,
we
can
open
up
our
database.
Client
I
am
using
PG
admin
for
postgres.
We
log
into
our
server
and
expand
our
database
under
the
schema.
Let's
locate
the
dim
underscore
product
underscore
category
table,
we
did
not
specify
a
schema,
so
by
default
this
table
will
be
stored
under
public
schema.