Execute Components in order with inputs
Use inputs when your Component needs to process data from upstream Components. Use dependencies (a special kind of input) when your Component only needs upstream Components to complete first without accessing their data, such as for setup tasks or configuration.
To gain a deeper understanding of these concepts, see the Components guide.
Looking to reference Components from other Flows? Use cross-Flow references.
Syntax
You can add inputs or dependencies to any Component type using the appropriate syntax:
Component type | Syntax | Location |
---|---|---|
YAML Components | inputs: / dependencies: field | Field in the Component configuration |
SQL Components | inputs[] / dependencies=[] parameter | Parameter in the config block |
Python Components | inputs[] / dependencies=[] parameter | Parameter in the Component decorator |
Read Components support a dependencies
field but not an inputs
field, since they don't process data from upstream Components.
YAML example
Define dependencies using the dependencies
field in your YAML configuration:
component:
read:
connection: read_gcs_lake
gcs:
path: ottos-expeditions/lakev0/generated/events/sales_store.parquet/year=
include:
- glob: "*/month=*/day=*/*.parquet"
dependencies:
- name: sales_website
flow: extract-load-duckdb
- name: sales_vendors
flow: extract-load-duckdb
SQL examples
Specify inputs and dependencies in the configuration block at the top of your SQL file:
{{
config(
type="transform",
inputs=[
ref('sales_vendors', flow='transform-duckdb'),
ref('read_sales_vendors', flow='extract-load-duckdb'),
]
)
}}
SELECT * FROM {{ ref('sales_vendors', flow='transform-duckdb') }}
Python examples
Add inputs and dependencies as parameters in your Component decorator. Tables referenced via ref()
must also be passed as function arguments. The implementation varies by Component type:
Transform
import ibis
from ascend.application.context import ComponentExecutionContext
from ascend.resources import ref, transform
@transform(
inputs=[
ref("ascenders", flow="transform-duckdb"),
ref("routes", flow="transform-duckdb"),
ref("telemetry", flow="transform-duckdb"),
]
)
def goats(
ascenders: ibis.Table,
routes: ibis.Table,
Incremental Python Read Component
import polars as pl
import pyarrow as pa
from ascend.application.context import ComponentExecutionContext
from ascend.resources import read, ref
@read(
strategy="incremental",
incremental_strategy="merge",
unique_key="id",
on_schema_change="sync_all_columns",
inputs=[ref("read_sales", flow="extract-load-duckdb")],
)
def read_inlinked(context: ComponentExecutionContext) -> pa.Table:
df = pl.read_parquet("gs://ascend-io-gcs-public/ottos-expeditions/lakev0/generated/events/inlinked.parquet/year=*/month=*/day=*/*.parquet")
Smart Python Read Component
from datetime import datetime
import pandas as pd
from ascend.application.context import ComponentExecutionContext
from ascend.common.filters import ListItem
from ascend.resources import CustomPythonPartitionedReader, ref
sales_1 = {
"sale_id": [1001, 1002, 1003],
"date": [datetime(2024, 1, 15), datetime(2024, 1, 16), datetime(2024, 1, 17)],
"product": ["Laptop", "Mouse", "Keyboard"],
"quantity": [2, 5, 3],
"customer_id": ["CUST001", "CUST002", "CUST003"],
}
Tasks
Tasks use the same syntax for dependencies and inputs as Transforms. Python Tasks follow the Python Transform syntax, while SQL Tasks follow the SQL Transform syntax.
Python Task
import ibis
from ascend.application.context import ComponentExecutionContext
from ascend.common.events import log
from ascend.resources import ref, task
@task(
dependencies=[
ref("social_media", flow="transform-duckdb"),
]
)
def task_respond_on_socials(social_media: ibis.Table, context: ComponentExecutionContext) -> None:
for i in range(1000):
log("Thank you for your comment!")
SQL Task
{{
config(
type="task",
dependencies=[ref("guides", flow="transform-duckdb")]
)
}}
SELECT 1 as id