Building Better Analytics Pipelines

Getting Started with Dagster ComponentsDagster Components introduces the new unified CLI tool, dg, to streamline project creation and management. The best place to get started with dg and components is with our [documentation](https://docs.dagster.io/guides/labs/components/).Once you have installed dg and created a project, you can quickly scaffold assets using components with dg. This component can then be customized using YAML, which makes adopting Dagster easier for a wider variety of users and use cases.For example, if you are a keen birdwatcher, you may have created an asset for a survey of birds for each year, which can lead to duplicated code:

‍

@dg.asset(kinds=["python"], group_name="raw_data")
def checklist_2020(context: dg.AssetExecutionContext):
    extracted_names, elapsed_times = download_and_extract_data(
        context, constants.CHECKLIST_2020
    )
    return dg.MaterializeResult(
        metadata={
            "names": extracted_names,
            "num_files": len(extracted_names),
            "elapsed_time": elapsed_times,
        },
    )


@dg.asset(kinds=["python"], group_name="raw_data")
def checklist_2023(context: dg.AssetExecutionContext):
    extracted_names, elapsed_times = download_and_extract_data(
        context, constants.CHECKLIST_2023
    )
    return dg.MaterializeResult(
        metadata={
            "names": extracted_names,
            "num_files": len(extracted_names),
            "elapsed_time": elapsed_times,
        },
    )

‍

‍

LLMs operate at their best when given constraints and structure. Dagster Components is built from the ground up to work alongside your favorite AI tools, from Copilot to Cursor, Claude Code to Cline. An MCP server will ship alongside Components to enable your favorite tools to seamlessly integrate with Dagster.

‍

‍

‍

@dg.asset(kinds=["python"], group_name="raw_data")
def checklist_2020(context: dg.AssetExecutionContext):
    extracted_names, elapsed_times = download_and_extract_data(
        context, constants.CHECKLIST_2020
    )
    return dg.MaterializeResult(
        metadata={
            "names": extracted_names,
            "num_files": len(extracted_names),
            "elapsed_time": elapsed_times,
        },
    )


@dg.asset(kinds=["python"], group_name="raw_data")
def checklist_2023(context: dg.AssetExecutionContext):
    extracted_names, elapsed_times = download_and_extract_data(
        context, constants.CHECKLIST_2023
    )
    return dg.MaterializeResult(
        metadata={
            "names": extracted_names,
            "num_files": len(extracted_names),
            "elapsed_time": elapsed_times,
        },
    )

‍

‍

‍

‍