NotebookUtils notebook draaien en orkestreren

Gebruik de hulpprogramma's voor notebooks om een notebook uit te voeren, meerdere notebooks parallel uit te voeren of een notebook met een waarde te beëindigen. Voer de volgende opdracht uit om een overzicht te krijgen van de beschikbare methoden:

notebookutils.notebook.help()

De volgende tabel bevat de beschikbare notebook-uitvoerings- en orkestratiemethoden.

Methode Signature Beschrijving
run run(path: str, timeout_seconds: int = 90, arguments: dict = None, workspace: str = ""): str Voert een notebook uit en retourneert de exitwaarde.
runMultiple runMultiple(dag: Any, config: dict = None): dict[str, dict[str, Any]] Hiermee worden meerdere notebooks gelijktijdig uitgevoerd met ondersteuning voor afhankelijkheidsrelaties.
validateDAG validateDAG(dag: Any): bool Valideert of een DAG-definitie juist is gestructureerd.
exit exit(value: str): None Sluit het huidige notitieblok af met een waarde.

Zie Notitieblokartefacten beheren voor CRUD-bewerkingen (maken, ophalen, bijwerken, verwijderen, lijst).

Opmerking

De config parameter in runMultiple() is alleen beschikbaar in Python. Scala en R bieden geen ondersteuning voor deze parameter.

Opmerking

Hulpprogramma's voor notebooks zijn niet van toepassing op Apache Spark-taakdefinities (SJD).

Verwijzen naar een notitieblok

De run() methode verwijst naar een notebook en retourneert de uitgangswaarde. U kunt geneste functie-aanroepen uitvoeren in een notebook of in een pijplijn, interactief. Het notebook waarnaar wordt verwezen draait op de Spark-pool van het notebook dat deze functie aanroept.

notebookutils.notebook.run("notebook name", <timeout_seconds>, <arguments>, <workspace>)

Voorbeeld:

notebookutils.notebook.run("Sample1", 90, {"input": 20 })

Retourwaarde

De run() methode retourneert de exacte tekenreeks die aan notebookutils.notebook.exit(value) is doorgegeven in het onderliggende notitieblok. Als exit() niet wordt aangeroepen in het kindnotitieblok, wordt een lege tekenreeks ("") geretourneerd.

Fabric-notebooks ondersteunen ook verwijzingen naar notebooks in werkruimten door de werkruimte-id op te geven.

notebookutils.notebook.run("Sample1", 90, {"input": 20 }, "fe0a6e2a-a909-4aa3-a698-0a651de790aa")

Open de koppeling naar de momentopname in de celuitvoer om de verwijzingsuitvoering te controleren. Met de momentopname worden de resultaten van de uitvoering vastgelegd en kunt u fouten opsporen in het notebook waarnaar wordt verwezen.

Schermopname van het resultaat van de verwijzingsuitvoering.

Schermopname van een voorbeeld van een momentopname.

Subnotitieblokken instellen voor het ontvangen van parameters

Wanneer u een subnotitieblok maakt dat wordt aangeroepen via run() of runMultiple(), stelt u een parametercel in zodat het notitieblok argumenten van het bovenliggende notitieblok kan ontvangen.

  1. Maak een codecel met standaardparameterwaarden.
  2. Markeer de cel als parametercel door Cel als parameter markeren in de notebook-gebruikersinterface te selecteren.
  3. Tijdens de uitvoering worden de waarden van de parametercel vervangen door de argumenten die zijn doorgegeven van het bovenliggende item.
# This cell should be marked as "parameters" cell
# Default values are overridden when the notebook is called
date = "2024-01-01"
region = "US"

Aanbeveling

Afsluitwaarden zijn altijd tekenreeksen. Als u een numerieke waarde in het bovenliggende notitieblok nodig hebt, converteert u het resultaat na het ophalen (bijvoorbeeld int(result)).

Overwegingen

  • Het referentienotebook voor meerdere werkruimten wordt ondersteund door runtimeversie 1.2 en hoger.
  • Als u de bestanden onder Notebook Resource gebruikt, gebruikt u notebookutils.nbResPath in het notitieblok waarnaar wordt verwezen om ervoor te zorgen dat deze naar dezelfde map wijst als de interactieve uitvoering.
  • Met een verwijzingsuitvoering kunnen child notebooks alleen worden uitgevoerd als ze hetzelfde lakehouse gebruiken als het oudernotebook, het lakehouse erven, of wanneer geen van beide een lakehouse definieert. De uitvoering wordt geblokkeerd als het kind een ander lakehouse opgeeft dan het ouder notitieblok. Als u deze controle wilt omzeilen, stelt u de argumenten useRootDefaultLakehouse: True in.
  • Roep notebookutils.notebook.exit(value) niet aan binnen een try-catch blok. De exit-aanroep heeft geen effect wanneer deze is opgenomen in de afhandeling van uitzonderingen.

Naslaginformatie over het parallel uitvoeren van meerdere notebooks

Gebruik notebookutils.notebook.runMultiple() om meerdere notebooks parallel of volgens een vooraf bepaalde topologische structuur uit te voeren. De API maakt gebruik van een multithreaded implementatie in een Spark-sessie, wat betekent dat notebook-uitvoeringen gedeelde rekenbronnen gebruiken.

Met notebookutils.notebook.runMultiple()kunt u het volgende doen:

  • Voer meerdere notebooks tegelijk uit, zonder te wachten tot elke notebook is voltooid.

  • Geef de afhankelijkheden en de volgorde van uitvoering voor uw notebooks op met behulp van een eenvoudige JSON-indeling.

  • Optimaliseer het gebruik van Spark-rekenresources en verlaag de kosten van uw Fabric-projecten.

  • Bekijk de momentopnamen van elke record van de notebookuitvoering in de uitvoer en debug of bewaak uw notebooktaken gemakkelijk.

  • Haal de uitvoerwaarde van elke uitgevoerde activiteit op en gebruik deze in vervolgactiviteiten.

Voer deze opdracht uit notebookutils.notebook.help("runMultiple") om meer voorbeelden en gebruiksgegevens weer te geven.

Een eenvoudige lijst met notebooks uitvoeren

In het volgende voorbeeld wordt een lijst met notebooks parallel uitgevoerd:

notebookutils.notebook.runMultiple(["NotebookSimple", "NotebookSimple2"])

Het uitvoerresultaat van het hoofdnotebook is als volgt:

Schermopname van een lijst met notitieblokken.

Retourwaarde

De runMultiple() methode retourneert een woordenlijst waarbij elke sleutel de naam van de activiteit is en elke waarde een woordenlijst is met de volgende sleutels:

  • exitVal: De tekenreeks die wordt geretourneerd door het aanroepen van exit() op het onderliggende notitieblok, of een lege tekenreeks als exit() niet is aangeroepen.
  • exception: een foutobject als de activiteit is mislukt of None als deze is geslaagd.

Notebooks uitvoeren met een DAG-structuur

In het volgende voorbeeld worden notebooks in een DAG-structuur uitgevoerd met behulp van notebookutils.notebook.runMultiple().

# run multiple notebooks with parameters
DAG = {
    "activities": [
        {
            "name": "Process_1", # activity name, must be unique
            "path": "NotebookSimple", # notebook item name
            "timeoutPerCellInSeconds": 90, # max timeout for each cell, default to 90 seconds
            "args": {"p1": "changed value", "p2": 100}, # notebook parameters
            "workspace":"WorkspaceName" # both name and id are supported
        },
        {
            "name": "Process_2",
            "path": "NotebookSimple2",
            "timeoutPerCellInSeconds": 120,
            "args": {"p1": "changed value 2", "p2": 200},
            "workspace":"id" # both name and id are supported
        },
        {
            "name": "Process_1.1",
            "path": "NotebookSimple2",
            "timeoutPerCellInSeconds": 120,
            "args": {"p1": "changed value 3", "p2": 300},
            "retry": 1,
            "retryIntervalInSeconds": 10,
            "dependencies": ["Process_1"] # list of activity names that this activity depends on
        }
    ],
    "timeoutInSeconds": 43200, # max timeout for the entire DAG, default to 12 hours
    "concurrency": 12 # max number of notebooks to run concurrently, default to 3x CPU cores, 0 means unlimited
}
notebookutils.notebook.runMultiple(DAG, {"displayDAGViaGraphviz": False})

Het uitvoerresultaat van het hoofdnotebook is als volgt:

Schermopname van de lijst met notebooks en hun parameters.

DAG-parameterreferentie

In de volgende tabel wordt elk veld beschreven dat u kunt gebruiken in de DAG-definitie:

Veld Niveau Verplicht Beschrijving
activities Wortel Ja Een lijst met activiteitsobjecten die de notebooks definiëren die moeten worden uitgevoerd.
timeoutInSeconds Wortel No Maximale time-out voor de hele DAG. De standaardwaarde is 43200 (12 uur).
concurrency Wortel No Maximum aantal notebooks dat gelijktijdig kan worden uitgevoerd. De standaardwaarde is 3 keer het aantal beschikbare CPU-kernen. Stel deze waarde expliciet in als u een strakkere controle nodig hebt of gebruik 0 voor onbeperkte gelijktijdigheid.
name Activity Ja Een unieke naam voor de activiteit. Wordt gebruikt om resultaten te identificeren en afhankelijkheden te definiëren.
path Activity Ja De naam of het pad van het notitieblokitem dat moet worden uitgevoerd.
timeoutPerCellInSeconds Activity No Maximale time-out voor elke cel in het onderliggende notitieblok. De standaardwaarde is 90 seconden.
args Activity No Een woordenlijst met parameters die moeten worden doorgegeven aan het onderliggende notitieblok.
workspace Activity No De naam of id van de werkruimte waarin het notitieblok zich bevindt. Standaard wordt het subnotitieblok uitgevoerd in dezelfde werkruimte als de aanroeper.
retry Activity No Aantal nieuwe pogingen als de activiteit mislukt. De standaardwaarde is 0.
retryIntervalInSeconds Activity No Wachttijd in seconden tussen nieuwe pogingen. De standaardwaarde is 0.
dependencies Activity No Een lijst met namen van activiteiten die moeten worden voltooid voordat deze activiteit wordt gestart.

Verwijzing naar uitstapwaarden tussen activiteiten

U kunt verwijzen naar de afsluitwaarde van een afhankelijkheidsactiviteit in het args veld met behulp van de @activity() expressie. Met dit patroon kunt u gegevens doorgeven tussen notebooks in een DAG.

DAG = {
    "activities": [
        {
            "name": "Extract",
            "path": "ExtractData",
            "timeoutPerCellInSeconds": 120,
            "args": {"source": "prod_db"}
        },
        {
            "name": "Transform",
            "path": "TransformData",
            "timeoutPerCellInSeconds": 180,
            "args": {
                "data_path": "@activity('Extract').exitValue()"
            },
            "dependencies": ["Extract"]
        }
    ]
}

results = notebookutils.notebook.runMultiple(DAG)

Aanbeveling

Gebruik de @activity('activity_name').exitValue() expressie in het args veld om resultaten van de ene activiteit door te geven aan een andere activiteit in een DAG.

Een dynamische DAG bouwen

U kunt DAG-structuren programmatisch genereren voor scenario's zoals bij fan-out-verwerking over meerdere partities:

def create_fan_out_dag(partitions):
    activities = []

    for partition in partitions:
        activities.append({
            "name": f"Process_{partition}",
            "path": "ProcessPartition",
            "timeoutPerCellInSeconds": 180,
            "args": {"partition": partition}
        })

    activities.append({
        "name": "Aggregate",
        "path": "AggregateResults",
        "timeoutPerCellInSeconds": 120,
        "dependencies": [f"Process_{p}" for p in partitions]
    })

    return {"activities": activities, "concurrency": 25}

partitions = ["2024-01", "2024-02", "2024-03", "2024-04"]
dag = create_fan_out_dag(partitions)

results = notebookutils.notebook.runMultiple(dag)

Een DAG valideren

Gebruik validateDAG() dit om te controleren of uw DAG-structuur geldig is voordat deze wordt uitgevoerd. Het detecteert problemen zoals dubbele activiteitsnamen, ontbrekende afhankelijkheden en circulaire verwijzingen.

notebookutils.notebook.validateDAG(DAG)

Retourwaarde

De validateDAG() methode retourneert True als de DAG-structuur geldig is of een uitzondering genereert als de validatie mislukt.

Aanbeveling

Roep validateDAG() altijd eerder aan in runMultiple() productiewerkstromen om structurele fouten vroeg te ondervangen.

RunMultiple-fouten afhandelen

De runMultiple() methode retourneert een woordenlijst waarbij elke sleutel de naam van de activiteit is en elke waarde een exitVal (tekenreeks) en een exception (foutobject) bevat.None U kunt gedeeltelijke resultaten inspecteren, zelfs wanneer sommige activiteiten mislukken:

from notebookutils.common.exceptions import RunMultipleFailedException

try:
    results = notebookutils.notebook.runMultiple(DAG)
except RunMultipleFailedException as ex:
    results = ex.result

for activity_name, result in results.items():
    if result["exception"]:
        print(f"{activity_name} failed: {result['exception']}")
    else:
        print(f"{activity_name} succeeded: {result['exitVal']}")

Overwegingen

  • De mate van parallelle uitvoering van meerdere notebooks is beperkt tot de totale beschikbare rekenresource van een Spark-sessie.
  • Het standaardaantal gelijktijdige notebooks is 3 keer het beschikbare aantal CPU-kernen. U kunt deze waarde aanpassen, maar overmatige parallelle uitvoering kan leiden tot stabiliteits- en prestatieproblemen vanwege een hoog rekenresourcegebruik. Als er problemen optreden, kunt u overwegen de notebooks op te splitsen in meerdere runMultiple aanroepen of de gelijktijdigheid te verlagen door het veld concurrency in de DAG-parameter bij te stellen.
  • De standaardtime-out voor de hele DAG is 12 uur en de standaardtime-out voor elke cel in een subnotitieblok is 90 seconden. U kunt de timeout wijzigen door de timeoutInSeconds en timeoutPerCellInSeconds in te stellen in de DAG-parameter.
  • Configureer retry en retryIntervalInSeconds voor activiteiten die kunnen mislukken vanwege problemen zoals netwerktime-outs of tijdelijke onbeschikbaarheid van de dienst.
  • Parallelle notebooks delen rekenresources binnen één Spark-sessie. Bewaak het resourcegebruik om geheugendruk en CPU-conflicten te voorkomen.

Een notitieblok afsluiten

De exit() methode sluit een notebook af en resulteert in een waarde. U kunt geneste functie-aanroepen uitvoeren in een notebook of in een pijplijn, interactief.

  • Wanneer u een exit() functie interactief aanroept vanuit een notebook, genereert het Fabric-notebook een uitzondering, slaat het uitvoeren van volgende cellen over en blijft de Spark-sessie actief.

  • Wanneer u een notebook in een pijplijn coördineert die een exit()-functie aanroept, geeft de notebookactiviteit een afsluitwaarde terug. Hiermee wordt de pijplijnuitvoering voltooid en wordt de Spark-sessie gestopt.

  • Wanneer u een exit()-functie aanroept in een notebook waarnaar wordt verwezen, stopt Fabric Spark de verdere executie van het betreffende notitieblok en worden de volgende cellen in het hoofdnotitieblok uitgevoerd, waarmee de run()-functie wordt aangeroepen. Bijvoorbeeld: Notebook1 heeft drie cellen en roept een exit() functie aan in de tweede cel. Notebook2 heeft vijf cellen en roept run(notebook1) aan in de derde cel. Wanneer u Notebook2 uitvoert, stopt Notebook1 in de tweede cel wanneer u de exit() functie bereikt. Notebook2 blijft zijn vierde en vijfde cellen uitvoeren.

notebookutils.notebook.exit("value string")

Retourgedrag

De exit() methode retourneert geen waarde. Hiermee wordt het huidige notebook beëindigd en wordt de opgegeven tekenreeks doorgegeven aan het aanroepende notebook of de pijplijn.

Opmerking

De exit() functie overschrijft de huidige celuitvoer. Als u wilt voorkomen dat de uitvoer van andere code-instructies verloren gaat, roept u notebookutils.notebook.exit() aan in een afzonderlijke cel.

Belangrijk

Roep notebookutils.notebook.exit() niet aan binnen een try-catch blok. De beëindiging heeft geen effect wanneer deze in uitzonderingsbehandeling is verpakt. De exit() aanroep moet zich op het hoogste niveau van uw code bevinden om correct te kunnen werken.

Voorbeeld:

Het notebook Sample1 heeft de volgende twee cellen:

  • Cel 1 definieert een invoerparameter met de standaardwaarde ingesteld op 10.

  • Cel 2 verlaat het notitieblok met invoer als uitvoerwaarde.

Schermopname met een voorbeeldnotitieblok van de uitstapfunctie.

U kunt het voorbeeld1 uitvoeren in een ander notebook met standaardwaarden:

exitVal = notebookutils.notebook.run("Sample1")
print (exitVal)

Uitvoer:

10

U kunt het voorbeeld1 uitvoeren in een ander notebook en de invoerwaarde instellen als 20:

exitVal = notebookutils.notebook.run("Sample1", 90, {"input": 20 })
print (exitVal)

Uitvoer:

20

Verwante inhoud

  • Last updated on