June 7, 2024 • 8 minute read •
Running Singer on Dagster
- Name
- Fraser Marlow
- Handle
- @frasermarlow
Dagster offers many embedded data movement and ingestion options, and we recently summarized them in this blog post. However, Dagster’s flexible framework makes it easy to run any ELT process.
In this post, we will look at how to set up a Singer ELT pipeline using a Singer Tap and Target, and how to leverage the components of Dagster’s orchestration system. This post should be valuable for data engineers with an existing Singer tap/target who want to control these from within Dagster.
We should also note that Singer Taps are supported by Airbyte and we offer a Meltano integration so you have a wide range of choices. The following tutorial is probably the most lightweight approach.
Pros and Cons: Running Singer Taps Within Dagster
By invoking the Tap and Target from within Dagster, we gain several advantages:
- You can make Singer an integral part of your scheduled pipelines.
- You can add metadata to your Singer taps and targets.
- Because Dagster pickles the JSON payload rather than pipe it directly, you can retry on failure, send the Tap payload to multiple targets, or debug a target without constantly hitting the upstream API.
Of course, there are a few drawbacks we should consider, namely that Singer is already not the most performant ingestion solution (due, in part, to the serialization/deserialization over stdIn/Out). With the addition of the Dagster steps, we increase latency and also introduce more storage requirements. You will want to consider this if you use Singer to migrate large amounts of data and if speed is a big concern.
Before We Start
This post assumes that you are somewhat familiar with Dagster’s core concepts. If you are not, we recommend the Dagster Essentials course over at Dagster University. We will not cover the basic installation of Dagster here, but plenty of guides exist, and it is covered in the Dagster University course.
We also assume you are familiar with Singer Taps and Targets and that you have set up and configured one of each and can run them with a command such as:
singer-tap –config config.json | singer-target
Overview of the Setup
Here is a breakdown of our setup:
- We will create a reusable Dagster
@resource
for the Tap and one for the Target. - We will create two assets that leverage the
@resource
and invoke the Tap and Target. - We will configure our Code Location's Definitions
- We will add the Assets to a job.
- We will watch the pipeline executing in Dagster's UI
We will create a dedicated code location for this exercise, but you can add the resources and assets to an existing code location. In this example, we will use a singer
folder:
(1) Home of our Definitions
(2) Creates two assets based on the @resource
, one for the Tap and one for the Target. We will also grab some metadata while we are at it.
(3) Sets up our two assets as a simple Dagster @job
(4) Defines a @resource
for the Tap and one for the Target
Setting Singer up on Dagster
Creating the Singer Resource
In the example below, we create a SingerTapResource
and a SingerTargetResource
class. These will be used to set the right parameters and invoke the command when the resource is invoked.
### singer/resources/__init__.py
from dagster import InitResourceContext
import subprocess
class SingerTapResource:
def __init__(self, context: InitResourceContext, tap_executable: str, config_path: str, catalog_path=None, state_path=None):
if check_config_path(config_path):
self.tap_executable = tap_executable
self.config_path = config_path
self.catalog_path = catalog_path
self.state_path = state_path
self.context = context
def run(self):
command = [self.tap_executable, '--config', self.config_path]
if self.catalog_path:
command.extend(['--catalog', self.catalog_path])
if self.state_path:
command.extend(['--state', self.state_path])
tap_result = subprocess.Popen(command, stdout=subprocess.PIPE, text=True)
output, errors = tap_result.communicate()
return output
class SingerTargetResource:
def __init__(self, target_executable: str, config_path: str, context: InitResourceContext):
if check_config_path(config_path):
self.target_executable = target_executable
self.config_path = config_path
self.context = context
def run(self, singer_tap_dataset):
command = [self.target_executable]
if self.config_path:
command.extend(['--config', self.config_path])
target_input = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
output, errors = target_input.communicate(input=singer_tap_dataset)
return output
Note that these objects make use of a helper function to ensure that the config.json
file is present:
from dagster import DagsterExecutionInterruptedError
import os
def check_config_path(config_path):
if not isinstance(config_path, str):
raise TypeError(f"config_path must be a string, got {type(config_path).__name__}")
if not os.path.isfile(config_path):
raise DagsterExecutionInterruptedError(f"Config file not found at {config_path}")
if not config_path.endswith('.json'):
raise DagsterExecutionInterruptedError(f"Config file at {config_path} is not a .json file")
return True
Now we can declare the @resource
:
from dagster import resource, InitResourceContext, Field, String
@resource(
config_schema={
"tap_executable": Field(String, description="Path to the tap executable"),
"config_path": Field(String, description="Path to the config.json file"),
"catalog_path": Field(String, is_required=False, description="Path to the catalog.json file"),
"state_path": Field(String, is_required=False, description="Path to the state.json file"),
}
)
def singer_tap_resource(init_context: InitResourceContext) -> SingerTapResource:
return SingerTapResource(
tap_executable=init_context.resource_config["tap_executable"],
config_path=init_context.resource_config["config_path"],
catalog_path=init_context.resource_config.get("catalog_path"),
state_path=init_context.resource_config.get("state_path"),
context=init_context
)
@resource(
config_schema={
"target_executable": Field(String, description="Path to the target executable"),
"config_path": Field(String, description="Path to the config.json file")
}
)
def singer_target_resource(init_context: InitResourceContext) -> SingerTargetResource:
return SingerTargetResource(
target_executable=init_context.resource_config["target_executable"],
config_path=init_context.resource_config["config_path"],
context=init_context
)
All told, our /resources/init.py
file looks as follows:
### singer/resources/__init__.py
from dagster import resource, InitResourceContext, Field, String, DagsterExecutionInterruptedError
import subprocess
import os
def check_config_path(config_path):
if not isinstance(config_path, str):
raise TypeError(f"config_path must be a string, got {type(config_path).__name__}")
if not os.path.isfile(config_path):
raise DagsterExecutionInterruptedError(f"Config file not found at {config_path}")
if not config_path.endswith('.json'):
raise DagsterExecutionInterruptedError(f"Config file at {config_path} is not a .json file")
return True
class SingerTapResource:
def __init__(self, context: InitResourceContext, tap_executable: str, config_path: str, catalog_path=None, state_path=None):
if check_config_path(config_path):
self.tap_executable = tap_executable
self.config_path = config_path
self.catalog_path = catalog_path
self.state_path = state_path
self.context = context
def run(self):
command = [self.tap_executable, '--config', self.config_path]
if self.catalog_path:
command.extend(['--catalog', self.catalog_path])
if self.state_path:
command.extend(['--state', self.state_path])
tap_result = subprocess.Popen(command, stdout=subprocess.PIPE, text=True)
output, errors = tap_result.communicate()
return output
class SingerTargetResource:
def __init__(self, target_executable: str, config_path: str, context: InitResourceContext):
if check_config_path(config_path):
self.target_executable = target_executable
self.config_path = config_path
self.context = context
def run(self, singer_tap_dataset):
command = [self.target_executable]
if self.config_path:
command.extend(['--config', self.config_path])
target_input = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True)
output, errors = target_input.communicate(input=singer_tap_dataset)
return output
@resource(
config_schema={
"tap_executable": Field(String, description="Path to the tap executable"),
"config_path": Field(String, description="Path to the config.json file"),
"catalog_path": Field(String, is_required=False, description="Path to the catalog.json file"),
"state_path": Field(String, is_required=False, description="Path to the state.json file"),
}
)
def singer_tap_resource(init_context: InitResourceContext) -> SingerTapResource:
return SingerTapResource(
tap_executable=init_context.resource_config["tap_executable"],
config_path=init_context.resource_config["config_path"],
catalog_path=init_context.resource_config.get("catalog_path"),
state_path=init_context.resource_config.get("state_path"),
context=init_context
)
@resource(
config_schema={
"target_executable": Field(String, description="Path to the target executable"),
"config_path": Field(String, description="Path to the config.json file")
}
)
def singer_target_resource(init_context: InitResourceContext) -> SingerTargetResource:
return SingerTargetResource(
target_executable=init_context.resource_config["target_executable"],
config_path=init_context.resource_config["config_path"],
context=init_context
)
Define the Dagster Assets
Tapping into the resources we have defined, we will now create an asset for our Tap and an asset for our Target (which takes the Tap as a dependency).
### singer/assets/singer.py
from dagster import asset, Output, AssetExecutionContext
@asset(required_resource_keys={"singer_tap"}, tags={"compute_kind": "tap"}, group_name="singer_taps")
def singer_tap_dataset(context: AssetExecutionContext) -> Output[str]:
"""
Call the Singer tap and retrieve the data
:return: JSON payload from the tap
:rtype: str
"""
singer_tap = context.resources.singer_tap
output = singer_tap.run() # Use the resource to run the tap
output_length = len(output)
num_streams, records_per_stream = count_streams_and_records(output)
return Output(value=output, metadata={"output_length": output_length})
@asset(required_resource_keys={"singer_target"}, tags={"compute_kind": "target"}, group_name="singer_targets")
def singer_target_dataset(context: AssetExecutionContext, singer_tap_dataset) -> Output[str]:
"""
Call the Singer target and submit the data
:return:
"""
singer_target = context.resources.singer_target
output = singer_target.run(singer_tap_dataset) # Use the resource to run the tap
return Output(value=output)
You will notice that we collect some metadata on the tap. In this case, it is a simple calculation of the length of the json object (metadata={"output_length": output_length}
).
We can build on this by introducing a simple count of the number of STREAMS
and RECORDS
used in the Tap. We can build this in as a separate function:
import json
from collections import defaultdict
def count_streams_and_records(json_payload):
# Split the payload into individual JSON objects
json_lines = json_payload.strip().split("\n")
# Initialize counters
streams_count = set()
records_count = defaultdict(int)
# Process each JSON object
for line in json_lines:
obj = json.loads(line)
# Count unique streams
if 'stream' in obj:
streams_count.add(obj['stream'])
# Count records per stream
if obj['type'] == 'RECORD' and 'stream' in obj:
records_count[obj['stream']] += 1
# Number of unique streams
num_streams = len(streams_count)
return num_streams, dict(records_count)
Let’s add this to our singer_tap_dataset
asset. Our singer.py
file now looks like this:
from dagster import asset, Output, AssetExecutionContext
import json
from collections import defaultdict
def count_streams_and_records(json_payload):
# Split the payload into individual JSON objects
json_lines = json_payload.strip().split("\n")
# Initialize counters
streams_count = set()
records_count = defaultdict(int)
# Process each JSON object
for line in json_lines:
obj = json.loads(line)
# Count unique streams
if 'stream' in obj:
streams_count.add(obj['stream'])
# Count records per stream
if obj['type'] == 'RECORD' and 'stream' in obj:
records_count[obj['stream']] += 1
# Number of unique streams
num_streams = len(streams_count)
return num_streams, dict(records_count)
@asset(required_resource_keys={"singer_tap"}, tags={"compute_kind": "tap"}, group_name="singer_taps")
def singer_tap_dataset(context: AssetExecutionContext) -> Output[str]:
"""
Call the Singer tap and retrieve the data
:return: JSON payload from the tap
:rtype: str
"""
singer_tap = context.resources.singer_tap
output = singer_tap.run() # Use the resource to run the tap
output_length = len(output)
num_streams, records_per_stream = count_streams_and_records(output)
return Output(value=output, metadata={"output_length": output_length,"count_of_streams": num_streams, "records_per_stream": records_per_stream})
@asset(required_resource_keys={"singer_target"}, tags={"compute_kind": "target"}, group_name="singer_targets")
def singer_target_dataset(context: AssetExecutionContext, singer_tap_dataset) -> Output[str]:
"""
Call the Singer target and submit the data
:return:
"""
singer_target = context.resources.singer_target
output = singer_target.run(singer_tap_dataset) # Use the resource to run the tap
return Output(value=output)
Configuring the Code Location Definitions
Next, we set up our code location definitions in singer/init.py
. You will need to replace the resources
parameters to match the tap (“my-tap
”) and the target (“my-target
”) that you plan on using. Note that it is a best practice to invoke both the Tap and the Target from within their respective virtual environments (see the Singer documentation for details).
### singer/__init__.py
from dagster import load_assets_from_modules, Definitions
from .assets import singer
from .resources import singer_tap_resource, singer_target_resource
singer_assets = load_assets_from_modules([singer])
all_jobs = [run_singer_job]
defs = Definitions(
assets=[*singer_assets],
jobs=all_jobs,
resources={
"singer_tap": singer_tap_resource.configured({
"tap_executable": "/Users/frasermarlow/.virtualenvs/my-tap/bin/my-tap",
"config_path": "/Users/frasermarlow/singer/my-tap-config/config.json",
"catalog_path": "/Users/frasermarlow/singer/my-tap-config/catalog.json"
}),
"singer_target": singer_target_resource.configured({
"target_executable": "/Users/frasermarlow/.virtualenvs/my-target/bin/target-csv",
"config_path": "/Users/frasermarlow/singer/my-target/config.json",
})
}
)
Set Up a Simple Job
Finally, to help make this easier to trigger, let’s wrap the two assets into a Dagster job:
### jobs/__init__.py
from dagster import AssetSelection, define_asset_job
all_assets = AssetSelection.all()
run_singer_job = define_asset_job(
name="run_singer_job",
selection=all_assets
)
And remember to import this into singer/__init__.py
:
from .jobs import run_singer_job
Running Our Singer Pipeline in the Dagster UI:
Let’s boot up Dagster and see our assets in the UI:
dagster dev -m ~/singer
Here we see the two assets rendered with their dependency modeled:
The assets are automatically added to our asset catalog…
… and, after a few materializations, the metadata is plotted for us, giving us nicer observability over the data flowing through our pipeline.
Metadata that is not in a numeric series, such as the names of the streams and the record counts, is still captured in the asset metadata.
Leveraging More of Dagster
The setup shown above might be adequate for your use case, but if you want to tap into more Dagster capabilities, you can now leverage advanced scheduling, sensors, data quality checks, and more.
Wrapping Up
Setting up a Singer Tap and Target is straightforward in Dagster and lets us weave these into our pipelines.
Using this combination can help you achieve better integration, observability, track metadata, and more.
We're always happy to hear your feedback, so please reach out to us! If you have any questions, ask them in the Dagster community Slack (join here!) or start a Github discussion. If you run into any bugs, let us know with a Github issue. And if you're interested in working with us, check out our open roles!
Follow us:
Orchestrate Unstructured Data Pipelines with Dagster and dlt
- Name
- Zaeem Athar
- Handle
- @zaeem
Parallel Computing on Dagster with Dask
- Name
- Odette Harary
- Handle
- @odette
Orchestrating dbt™ with Dagster
- Name
- Rex Ledesma
- Handle
- @_rexledesma
- Name
- Sandy Ryza
- Handle
- @s_ryz