Return a writer that writes the error rows to a parquet file under error_rows_dir.
wiring.build_file_writer(
context,
error_rows_key,
error_rows_dir,
)
Use it in a direct invocation, where build_io_manager_writer raises because there is no step. The file is at get_error_rows_path(error_rows_key, error_rows_dir, context.partition_key).
Parameters
context: dg.AssetExecutionContext-
The context the asset was invoked with. The writer reads only its partition key, so a context from dg.build_asset_context() works unbound. Without a partition key, the rows go to the unpartitioned path.
error_rows_key: dg.AssetKey-
The key the writer writes under, such as the one get_error_rows_key returns.
error_rows_dir: UPath | Path | str-
The directory the file goes under. Any path
UPath accepts.
Returns
ErrorRowsWriter-
A writer that takes the error rows and returns the file’s path as a string.
Raises
dg.DagsterInvariantViolationError-
The context has a partition range and no partition key. The error rows do not record which partition each row came from, so they cannot be split into one file per partition.
Examples
import tempfile
import dagster as dg
import dagster_dataframely as dd
from dagster_dataframely.data import OrdersSchema, marketplace_orders
error_rows_dir = tempfile.mkdtemp()
write = dd.wiring.build_file_writer(
dg.build_asset_context(),
dd.wiring.get_error_rows_key(dg.AssetKey("orders")),
error_rows_dir,
)
_, failure = OrdersSchema.filter(marketplace_orders, cast=False)
write(dd.wiring.get_error_rows(OrdersSchema, failure)).removeprefix(error_rows_dir)
'/orders__error_rows.parquet'