Are people calling the Nvidia cosmos paper the next "attention is all you need" but for physical AI systems?
Sean Lopp
@loppsean.bsky.social
598 followers 322 following 70 posts
Dagster, RStudio, Looker. Dad. Bikes. Coffee.
Dagster, RStudio, Looker. Dad. Bikes. Coffee.
![@dg.asset(
deps=[csv_external_asset],
automation_condition=dg.AutomationCondition.on_cron("* * * * *"),
)
def orders(context: dg.AssetExecutionContext, csv_storage: CsvStorageResource) -> None:
orders = csv_storage.read_data('orders_raw.csv')
context.log.info(f"Creating asset one with data: {orders.head()}")
# directly write csv to storage
csv_storage.write_data(orders, 'orders.csv')
return dg.MaterializeResult(
metadata={
"dagster/row_count": dg.MetadataValue.int(len(orders)),
"preview": dg.MetadataValue.md(orders.head().to_markdown()),
"dagster/column_schema": create_table_schema_metadata_from_dataframe(orders)
}
)
@dg.asset_check(asset="orders")
def orders_increasing_or_equal(context: dg.AssetCheckExecutionContext) -> dg.AssetCheckResult:
histotical_metadata = context.instance.fetch_materializations(
dg.AssetRecordsFilter(
asset_key=dg.AssetKey("orders"),
),
limit=2
)
if len(histotical_metadata) < 2:
return dg.AssetCheckResult(passed=True)
row_count_values = [
record.asset_event.metadata.get("dagster/row_count") for record in histotical_metadata[0]
]
latest_row_count = row_count_values[0]
previous_row_count = row_count_values[1]
return dg.AssetCheckResult(
passed=latest_row_count>=previous_row_count,
metadata={"latest_row_count": latest_row_count, "previous_row_count": previous_row_count}
)](https://cdn.bsky.app/img/feed_thumbnail/plain/did:plc:lc5jzrr425fyah724df3z5ik/bafkreic6g4dx2u76s5odyhy3rnzuszybc3ara4kuvxvud7yo2umvcya33i)
![# define a resource that can read and write data
class CsvStorageResource(dg.ConfigurableResource):
base_dir: str = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"data",
)
def read_data(self, csv_file):
return pd.read_csv(os.path.join(self.base_dir, csv_file) )
def write_data(self, df, csv_file):
return df.to_csv(os.path.join(self.base_dir, csv_file), index=False )
# define an asset that represents an external CSV file
csv_external_asset = dg.AssetSpec(
key = "csv_external_asset",
description="A CSV file that is external to Dagster. If you have issues call 866-5309",
metadata={"file_path": dg.MetadataValue.path("data/orders_raw.csv"),
},
owners=["team:Upstream_Ops", "upstream_ops_person@company.com"],
kinds=["file","csv"],
)
@dg.asset(
deps=[csv_external_asset],
automation_condition=dg.AutomationCondition.on_cron("* * * * *"),
)
def orders(context: dg.AssetExecutionContext, csv_storage: CsvStorageResource) -> None:
orders = csv_storage.read_data('orders_raw.csv')
context.log.info(f"Creating asset one with data: {orders.head()}")
# directly write csv to storage
csv_storage.write_data(orders, 'orders.csv')
return dg.MaterializeResult(
metadata={
"dagster/row_count": dg.MetadataValue.int(len(orders)),
"preview": dg.MetadataValue.md(orders.head().to_markdown()),
"dagster/column_schema": create_table_schema_metadata_from_dataframe(orders)
}
)
@dg.asset(
deps=["orders_summary"],
automation_condition=dg.AutomationCondition.on_cron("*/10 * * * *")
)
def asset_three(context: dg.AssetExecutionContext) -> None:
context.log.info("Creating asset three")
# define the resources to use different folders for different "environments"
resource_defs = {
"DEV":{"csv_storage": CsvStorageResource(base_dir="/Users/christian/code/30_days_of_dagster/data…](https://cdn.bsky.app/img/feed_thumbnail/plain/did:plc:lc5jzrr425fyah724df3z5ik/bafkreiguofxoye2lcnttae5fc42eclrqicqhzvybxszydzn74oyml33fl4)
