diff --git a/docs/content/api/modules.json.gz b/docs/content/api/modules.json.gz
index fc2b8e7acde70..9f2fc1eea29bc 100644
Binary files a/docs/content/api/modules.json.gz and b/docs/content/api/modules.json.gz differ
diff --git a/docs/content/api/searchindex.json.gz b/docs/content/api/searchindex.json.gz
index 344fec5d97d9d..572a62a18d818 100644
Binary files a/docs/content/api/searchindex.json.gz and b/docs/content/api/searchindex.json.gz differ
diff --git a/docs/content/api/sections.json.gz b/docs/content/api/sections.json.gz
index 2719a26b1c11a..48ef220b5cf75 100644
Binary files a/docs/content/api/sections.json.gz and b/docs/content/api/sections.json.gz differ
diff --git a/docs/content/concepts/dagster-pipes/aws-glue.mdx b/docs/content/concepts/dagster-pipes/aws-glue.mdx
index 6c4c584597ff0..ed89a0d8a3a03 100644
--- a/docs/content/concepts/dagster-pipes/aws-glue.mdx
+++ b/docs/content/concepts/dagster-pipes/aws-glue.mdx
@@ -8,7 +8,6 @@ description: "Learn to integrate Dagster Pipes with AWS Glue to launch external
This tutorial gives a short overview on how to use [Dagster Pipes](/concepts/dagster-pipes) with [AWS Glue](https://aws.amazon.com/glue/).
The [dagster-aws](/\_apidocs/libraries/dagster-aws) integration library provides the resource which can be used to launch AWS Glue jobs from Dagster assets and ops. Dagster can receive regular events like logs, asset checks, or asset materializations from jobs launched with this client. Using it requires minimal code changes on the job side.
-
---
## Prerequisites
@@ -46,24 +45,26 @@ Call `open_dagster_pipes` in the Glue job script to create a context that can be
import boto3
from dagster_pipes import (
PipesCliArgsParamsLoader,
+ PipesDefaultMessageWriter,
PipesS3ContextLoader,
- PipesS3MessageWriter,
open_dagster_pipes,
)
client = boto3.client("s3")
context_loader = PipesS3ContextLoader(client)
-message_writer = PipesS3MessageWriter(client)
params_loader = PipesCliArgsParamsLoader()
def main():
with open_dagster_pipes(
context_loader=context_loader,
- message_writer=message_writer,
params_loader=params_loader,
) as pipes:
pipes.log.info("Hello from AWS Glue job!")
+ pipes.report_asset_materialization(
+ metadata={"some_metric": {"raw_value": 0, "type": "int"}},
+ data_version="alpha",
+ )
if __name__ == "__main__":
@@ -107,7 +108,7 @@ Next, add the `PipesGlueClient` resource to your project's `) created by the Glue job to receive Dagster events and will forward this stream to `stdout`. If this is undesired, instead, the client can be configured to use , and the Glue job can use .
+
+
---
## Related
diff --git a/docs/next/public/objects.inv b/docs/next/public/objects.inv
index 95e299330b810..d6140f2e78adc 100644
Binary files a/docs/next/public/objects.inv and b/docs/next/public/objects.inv differ
diff --git a/docs/sphinx/sections/api/apidocs/libraries/dagster-aws.rst b/docs/sphinx/sections/api/apidocs/libraries/dagster-aws.rst
index a0b72ecb195f0..3de17d8fb855d 100644
--- a/docs/sphinx/sections/api/apidocs/libraries/dagster-aws.rst
+++ b/docs/sphinx/sections/api/apidocs/libraries/dagster-aws.rst
@@ -98,9 +98,28 @@ Resources which surface SecretsManager secrets for use in Dagster resources and
Pipes
--------------
+Context Injectors
+^^^^^^^^^^^^^^^^^
+
+.. autoclass:: dagster_aws.pipes.PipesS3ContextInjector
+
+.. autoclass:: dagster_aws.pipes.PipesLambdaEventContextInjector
+
+Message Readers
+^^^^^^^^^^^^^^^
+
+.. autoclass:: dagster_aws.pipes.PipesS3MessageReader
+
+.. autoclass:: dagster_aws.pipes.PipesCloudWatchMessageReader
+ :members: consume_cloudwatch_logs
+
+Clients
+^^^^^^^
+
.. autoclass:: dagster_aws.pipes.PipesLambdaClient
.. autoclass:: dagster_aws.pipes.PipesGlueClient
+ :members: run
Legacy
--------
diff --git a/examples/docs_snippets/docs_snippets/guides/dagster/dagster_pipes/glue/dagster_code.py b/examples/docs_snippets/docs_snippets/guides/dagster/dagster_pipes/glue/dagster_code.py
index 0dc0fb168ae56..24345a471d583 100644
--- a/examples/docs_snippets/docs_snippets/guides/dagster/dagster_pipes/glue/dagster_code.py
+++ b/examples/docs_snippets/docs_snippets/guides/dagster/dagster_pipes/glue/dagster_code.py
@@ -24,7 +24,7 @@ def glue_pipes_asset(
# start_definitions_marker
from dagster import Definitions # noqa
-from dagster_aws.pipes import PipesGlueContextInjector, PipesS3MessageReader
+from dagster_aws.pipes import PipesGlueContextInjector
bucket = os.environ["DAGSTER_GLUE_S3_CONTEXT_BUCKET"]
@@ -39,9 +39,6 @@ def glue_pipes_asset(
client=boto3.client("s3"),
bucket=bucket,
),
- message_reader=PipesS3MessageReader(
- client=boto3.client("s3"), bucket=bucket
- ),
)
},
)
diff --git a/examples/docs_snippets/docs_snippets/guides/dagster/dagster_pipes/glue/glue_script.py b/examples/docs_snippets/docs_snippets/guides/dagster/dagster_pipes/glue/glue_script.py
index 8402f980f1866..df7c61a10fd4d 100644
--- a/examples/docs_snippets/docs_snippets/guides/dagster/dagster_pipes/glue/glue_script.py
+++ b/examples/docs_snippets/docs_snippets/guides/dagster/dagster_pipes/glue/glue_script.py
@@ -2,23 +2,24 @@
from dagster_pipes import (
PipesCliArgsParamsLoader,
PipesS3ContextLoader,
- PipesS3MessageWriter,
open_dagster_pipes,
)
client = boto3.client("s3")
context_loader = PipesS3ContextLoader(client)
-message_writer = PipesS3MessageWriter(client)
params_loader = PipesCliArgsParamsLoader()
def main():
with open_dagster_pipes(
context_loader=context_loader,
- message_writer=message_writer,
params_loader=params_loader,
) as pipes:
pipes.log.info("Hello from AWS Glue job!")
+ pipes.report_asset_materialization(
+ metadata={"some_metric": {"raw_value": 0, "type": "int"}},
+ data_version="alpha",
+ )
if __name__ == "__main__":
diff --git a/python_modules/libraries/dagster-aws/dagster_aws/pipes.py b/python_modules/libraries/dagster-aws/dagster_aws/pipes.py
index 6bd7b35bcf6f3..4c4a6716f6db3 100644
--- a/python_modules/libraries/dagster-aws/dagster_aws/pipes.py
+++ b/python_modules/libraries/dagster-aws/dagster_aws/pipes.py
@@ -181,7 +181,7 @@ class PipesCloudWatchMessageReader(PipesMessageReader):
def __init__(self, client: Optional[boto3.client] = None):
"""Args:
- client (boto3.client): boto3 CloudWatch client.
+ client (boto3.client): boto3 CloudWatch client.
"""
self.client = client or boto3.client("logs")
@@ -204,6 +204,19 @@ def consume_cloudwatch_logs(
start_time: Optional[int] = None,
end_time: Optional[int] = None,
) -> None:
+ """Reads logs from AWS CloudWatch and forwards them to Dagster for events extraction and logging.
+
+ Args:
+ log_group (str): CloudWatch log group name
+ log_stream (str): CLoudWatch log stream name
+ start_time (Optional[int]): The start of the time range, expressed as the number of
+ milliseconds after Jan 1, 1970 00:00:00 UTC. Events with a timestamp equal to this
+ time or later than this time are included.
+ Events with a timestamp earlier than this time are not included.
+ end_time (Optional[int]): The end of the time range, expressed as the number of
+ milliseconds after Jan 1, 1970 00:00:00 UTC. Events with a timestamp equal to or
+ later than this time are not included.
+ """
handler = check.not_none(
self._handler, "Can only consume logs within context manager scope."
)
@@ -216,7 +229,7 @@ def consume_cloudwatch_logs(
extract_message_or_forward_to_stdout(handler, log_line)
def no_messages_debug_text(self) -> str:
- return "Attempted to read messages by extracting them from the tail of CloudWatch logs directly."
+ return "Attempted to read messages by extracting them from CloudWatch logs directly."
def _get_all_cloudwatch_events(
self,
@@ -247,6 +260,9 @@ def _get_all_cloudwatch_events(
class PipesLambdaEventContextInjector(PipesEnvContextInjector):
+ """Injects context via AWS Lambda event input.
+ Should be paired with :py:class`~dagster_pipes.PipesMappingParamsLoader` on the Lambda side.
+ """
def no_messages_debug_text(self) -> str:
return "Attempted to inject context via the lambda event input."
@@ -255,7 +271,7 @@ class PipesLambdaClient(PipesClient, TreatAsResourceParam):
"""A pipes client for invoking AWS lambda.
By default context is injected via the lambda input event and messages are parsed out of the
- 4k tail of logs. S3
+ 4k tail of logs.
Args:
client (boto3.client): The boto lambda client used to call invoke.
@@ -351,7 +367,11 @@ class PipesGlueClient(PipesClient, TreatAsResourceParam):
context_injector (Optional[PipesContextInjector]): A context injector to use to inject
context into the Glue job, for example, :py:class:`PipesGlueContextInjector`.
message_reader (Optional[PipesMessageReader]): A message reader to use to read messages
- from the glue job run. Defaults to :py:class:`PipesGlueLogsMessageReader`.
+ from the glue job run. Defaults to :py:class:`PipesGlueLogsMessageReader`.
+ When provided with :py:class:`PipesCloudWatchMessageReader`,
+ it will be used to recieve logs and events from the `.../output/`
+ CloudWatch log stream created by AWS Glue. Note that AWS Glue routes both
+ `stderr` and `stdout` from the main job process into this LogStream.
client (Optional[boto3.client]): The boto Glue client used to launch the Glue job
"""