(pipeline() .source(read_csv("visits.csv")) .pipe(enrich) .filter(lambda r: r["country"] == "US") .sink(write_jsonl("us_visits.jsonl")) ).run() juq470 provides a catch operator to isolate faulty rows without stopping the whole pipeline:
def capitalize_name(row): row["name"] = row["name"].title() return row juq470
enrich = lambda src: src.map(enrich_with_geo) Now enrich can be inserted anywhere in a pipeline: (pipeline()
Explore more stories
Agentic BI: How We 10x Time to Insight by Baking Claude Into Data Workflows
The classic BI pipeline is disrupted beyond repair. By integrating Claude and moving to an agentic BI workflow, we’ve shifted from manual reporting to high-level context engineering.
Agentic Engineering on Databricks: Toolkits, Setup, and Best Practices
AI coding on Databricks got a serious upgrade with the arrival of new toolkits. Here's how to choose the right one, set up your agentic workflow, and ship faster but without cutting corners.
Migrating Modern Data Platforms in 2026: Methodology, Automation, Governance
Code translation is only part of the story. This guide on migrating data platforms covers the full ecosystem: methodology, automation, governance, and ways to safely cut conversion time with AI by up to 70%.