Dodawanie zadań do prac w pakietach zasobów Databricks

Ta strona zawiera informacje o sposobie definiowania zadań roboczych w Pakietach Zasobów Databricks. Aby uzyskać informacje o zadaniach, zobacz sekcję Konfigurowanie i edytowanie zadań w Lakeflow Jobs.

Konfigurowanie zadań

Zdefiniuj zadania dla pracy w pakiecie w kluczu tasks definicji zadania. Przykłady konfiguracji zadań dostępnych typów zadań znajduje się w sekcji Ustawienia zadania . Aby uzyskać informacje na temat definiowania zadania w pakiecie, zobacz job (Zadanie).

Aby ustawić wartości zadań, większość typów zadań ma parametry specyficzne dla zadania, ale można również zdefiniować parametry zadania , które są przekazywane do zadań podrzędnych. Odwołania do wartości dynamicznych są obsługiwane dla parametrów zadania, które umożliwiają przekazywanie wartości specyficznych dla uruchomienia zadania między zadaniami. Aby uzyskać pełne informacje na temat przekazywania wartości zadań według typu zadania, zobacz Szczegóły według typu zadania.

Możesz również zastąpić ustawienia zadania ogólnego zadania ustawieniami docelowego obszaru roboczego. Zobacz Nadpisanie z użyciem ustawień docelowych.

Poniższa przykładowa konfiguracja definiuje zadanie z dwoma zadaniami opartymi na notatniku i przekazuje wartość z pierwszego zadania do drugiego.

resources:
  jobs:
    pass_task_values_job:
      name: pass_task_values_job
      tasks:
        # Output task
        - task_key: output_value
          notebook_task:
            notebook_path: ../src/output_notebook.ipynb

        # Input task
        - task_key: input_value
          depends_on:
            - task_key: output_value
          notebook_task:
            notebook_path: ../src/input_notebook.ipynb
            base_parameters:
              received_message: '{{tasks.output_value.values.message}}'

Zawiera output_notebook.ipynb następujący kod, który ustawia wartość zadania dla klucza message:

# Databricks notebook source
# This first task sets a simple output value.

message = "Hello from the first task"

# Set the message to be used by other tasks
dbutils.jobs.taskValues.set(key="message", value=message)

print(f"Produced message: {message}")

Element input_notebook.ipynb pobiera wartość parametru received_message, który został ustawiony w konfiguracji zadania:

# This notebook receives the message as a parameter.

dbutils.widgets.text("received_message", "")
received_message = dbutils.widgets.get("received_message")

print(f"Received message: {received_message}")

Ustawienia zadania

Ta sekcja zawiera ustawienia i przykłady dla każdego typu zadania.

Zadanie notatnika cleanroom

Zadanie notatnika czystych pomieszczeń uruchamia notatnik czystych pomieszczeń, gdy pole clean_rooms_notebook_task jest obecne. Aby uzyskać informacje na temat czystych pomieszczeń, zobacz Co to jest usługa Azure Databricks Clean Rooms?.

Następujące klucze są dostępne dla zadania notesu pomieszczeń czystych. Aby uzyskać odpowiednią definicję obiektu interfejsu API REST, zobacz clean_rooms_notebook_task.

Warunek zadania

condition_task umożliwia dodanie zadania z logiką warunkową if/else do zadania. Zadanie ocenia warunek, który może służyć do kontrolowania wykonywania innych zadań. Zadanie warunkowe nie wymaga klastra do wykonania i nie obsługuje ponownych prób ani powiadomień. Aby uzyskać więcej informacji na temat zadania warunkowego if/else, zobacz Dodawanie logiki rozgałęziania do zadania za pomocą zadania If/else.

Następujące klucze są dostępne dla zadania warunkowego. Aby uzyskać odpowiednią definicję obiektu interfejsu API REST, zobacz condition_task.

Przykłady

Poniższy przykład zawiera zadanie warunkowe i zadanie notatnika, które jest wykonywane tylko wtedy, gdy liczba napraw zadań jest mniejsza niż 5.

resources:
  jobs:
    my-job:
      name: my-job
      tasks:
        - task_key: condition_task
          condition_task:
            op: LESS_THAN
            left: '{{job.repair_count}}'
            right: '5'
        - task_key: notebook_task
          depends_on:
            - task_key: condition_task
              outcome: 'true'
          notebook_task:
            notebook_path: ../src/notebook.ipynb

Zadanie pulpitu nawigacyjnego

Używasz tego zadania, aby odświeżyć pulpit nawigacyjny i wysłać zrzut ekranu do subskrybentów. Aby uzyskać więcej informacji na temat pulpitów nawigacyjnych w pakietach, zobacz pulpit nawigacyjny.

Następujące klucze są dostępne dla zadania pulpitu nawigacyjnego. Aby uzyskać odpowiednią definicję obiektu interfejsu API REST, zobacz dashboard_task.

Przykłady

Poniższy przykład dodaje zadanie związane z pulpitem nawigacyjnym do projektu. Po uruchomieniu zadania pulpit nawigacyjny z określonym identyfikatorem zostanie odświeżony.

resources:
  jobs:
    my-dashboard-job:
      name: my-dashboard-job
      tasks:
        - task_key: my-dashboard-task
          dashboard_task:
            dashboard_id: 11111111-1111-1111-1111-111111111111

dbt - zadanie

Używasz tego zadania do uruchamiania jednego lub więcej poleceń dbt. Aby uzyskać więcej informacji na temat dbt, zobacz Connect to dbt Cloud.

Następujące klucze są dostępne dla zadania dbt. Aby uzyskać odpowiednią definicję obiektu interfejsu API REST, zobacz dbt_task.

Przykłady

Poniższy przykład dodaje zadanie dbt do zlecenia. To zadanie dbt używa określonego magazynu SQL w celu uruchomienia określonych poleceń dbt.

Aby uzyskać identyfikator usługi SQL Warehouse, otwórz stronę ustawień usługi SQL Warehouse, a następnie skopiuj identyfikator znaleziony w nawiasach po nazwie magazynu w polu Nazwa na karcie Przegląd .

resources:
  jobs:
    my-dbt-job:
      name: my-dbt-job
      tasks:
        - task_key: my-dbt-task
          dbt_task:
            commands:
              - 'dbt deps'
              - 'dbt seed'
              - 'dbt run'
            project_directory: /Users/someone@example.com/Testing
            warehouse_id: 1a111111a1111aa1
          libraries:
            - pypi:
                package: 'dbt-databricks>=1.0.0,<2.0.0'

Dla każdego zadania

for_each_task umożliwia ci dodanie zadania za pomocą pętli 'for each' do projektu. Zadanie wykonuje zagnieżdżone zadanie dla każdych dostarczonych danych wejściowych. Aby uzyskać więcej informacji na temat for_each_task, zobacz Używanie zadania For each do uruchamiania innego zadania w pętli.

Następujące klucze są dostępne dla elementu for_each_task. Aby uzyskać odpowiednią definicję obiektu interfejsu API REST, zobacz for_each_task.

Przykłady

Poniższy przykład dodaje for_each_task do zadania, gdzie iteruje po wartościach innego zadania i je przetwarza.

resources:
  jobs:
    my_job:
      name: my_job
      tasks:
        - task_key: generate_countries_list
          notebook_task:
            notebook_path: ../src/generate_countries_list.ipnyb
        - task_key: process_countries
          depends_on:
            - task_key: generate_countries_list
          for_each_task:
            inputs: '{{tasks.generate_countries_list.values.countries}}'
            task:
              task_key: process_countries_iteration
              notebook_task:
                notebook_path: ../src/process_countries_notebook.ipnyb

Zadanie JAR

To zadanie służy do uruchamiania pliku JAR. Możesz korzystać z lokalnych bibliotek JAR lub tych w przestrzeni roboczej, woluminie katalogu Unity lub w zewnętrznej lokalizacji w magazynie chmurowym. Zobacz Plik JAR (Java lub Scala).

Aby uzyskać szczegółowe informacje na temat kompilowania i wdrażania plików JAR języka Scala w klastrze z obsługą wykazu aparatu Unity w trybie dostępu standardowego, zobacz Samouczek: uruchamianie kodu Scala na bezserwerowych obliczeniach.

Następujące klucze są dostępne dla zadania JAR. Aby uzyskać odpowiednią definicję obiektu interfejsu API REST, zobacz jar_task.

Przykłady

Poniższy przykład dodaje zadanie JAR do zadania. Ścieżka do pliku JAR prowadzi do lokalizacji woluminu.

resources:
  jobs:
    my-jar-job:
      name: my-jar-job
      tasks:
        - task_key: my-jar-task
          spark_jar_task:
            main_class_name: org.example.com.Main
          libraries:
            - jar: /Volumes/main/default/my-volume/my-project-0.1.0-SNAPSHOT.jar

Zadanie notatnika

To zadanie służy do uruchamiania notatnika. Zobacz Zadanie notatnika dotyczące prac.

Następujące klucze są dostępne dla zadania notatnika. Aby uzyskać odpowiednią definicję obiektu interfejsu API REST, zobacz notebook_task.

Przykłady

Poniższy przykład dodaje zadanie notatnika do pracy i ustawia parametr pracy o nazwie my_job_run_id. Ścieżka do notebooka jest relatywna do pliku konfiguracji, w którym określono to zadanie. Zadanie pobiera notatnik z jego wdrożonej lokalizacji w obszarze roboczym usługi Azure Databricks.

resources:
  jobs:
    my-notebook-job:
      name: my-notebook-job
      tasks:
        - task_key: my-notebook-task
          notebook_task:
            notebook_path: ./my-notebook.ipynb
      parameters:
        - name: my_job_run_id
          default: '{{job.run_id}}'

Zadanie potoku

To zadanie służy do uruchamiania potoku. Zobacz Lakeflow Spark Declarative Pipelines.

Następujące kluczowe elementy są dostępne dla zadania potokowego. Aby uzyskać odpowiednią definicję obiektu interfejsu API REST, zobacz pipeline_task.

Przykłady

Poniższy przykład dodaje zadanie potoku do zadania. To zadanie uruchamia określoną linię przetwarzania.

resources:
  jobs:
    my-pipeline-job:
      name: my-pipeline-job
      tasks:
        - task_key: my-pipeline-task
          pipeline_task:
            pipeline_id: 11111111-1111-1111-1111-111111111111

Zadanie usługi Power BI

To zadanie służy do wyzwalania odświeżania modelu semantycznego usługi Power BI (dawniej nazywanego zestawem danych).

Następujące klucze są dostępne dla zadania usługi Power BI. Aby uzyskać odpowiednią definicję obiektu interfejsu API REST, zobacz power_bi_task.

Przykłady

W poniższym przykładzie zdefiniowano zadanie usługi Power BI, które określa połączenie, model usługi Power BI do zaktualizowania oraz tabelę usługi Databricks do wyeksportowania.

resources:
  jobs:
    my_job:
      name: my_job
      tasks:
        - task_key: power_bi_task
          power_bi_task:
            connection_resource_name: 'connection_name'
            power_bi_model:
              workspace_name: 'workspace_name'
              model_name: 'model_name'
              storage_mode: 'DIRECT_QUERY'
              authentication_method: 'OAUTH'
              overwrite_existing: false
            refresh_after_update: false
            tables:
              - catalog: 'main'
                schema: 'tpch'
                name: 'customers'
                storage_mode: 'DIRECT_QUERY'
            warehouse_id: '1a111111a1111aa1'

Zadanie skryptu języka Python

To zadanie służy do uruchamiania pliku w języku Python.

Następujące klucze są dostępne dla zadania skryptu języka Python. Aby uzyskać odpowiednią definicję obiektu interfejsu API REST, zobacz python_task.

Przykłady

Poniższy przykład dodaje zadanie skryptu w języku Python do projektu. Ścieżka pliku języka Python do wdrożenia jest względna względem pliku konfiguracji, w którym to zadanie jest zadeklarowane. Zadanie pobiera plik języka Python z wdrożonej lokalizacji w obszarze roboczym usługi Azure Databricks.

resources:
  jobs:
    my-python-script-job:
      name: my-python-script-job

      tasks:
        - task_key: my-python-script-task
          spark_python_task:
            python_file: ./my-script.py

Zadanie koła języka Python

To zadanie służy do uruchamiania koła języka Python. Zobacz Tworzenie pliku wheel języka Python przy użyciu pakietów zasobów usługi Databricks.

Następujące klucze są dostępne dla zadania Python wheel. Aby uzyskać odpowiednią definicję obiektu interfejsu API REST, zobacz python_wheel_task.

Przykłady

W poniższym przykładzie do zadania dodano polecenie tworzenia pakietu Wheel dla języka Python. Ścieżka pliku koła Python do wdrożenia jest względna wobec pliku konfiguracyjnego, w którym to zadanie jest zadeklarowane. Zobacz zależności biblioteki Databricks Asset Bundles.

resources:
  jobs:
    my-python-wheel-job:
      name: my-python-wheel-job
      tasks:
        - task_key: my-python-wheel-task
          python_wheel_task:
            entry_point: run
            package_name: my_package
          libraries:
            - whl: ./my_package/dist/my_package-*.whl

Uruchomienie zadania roboczego

Używasz tego zadania do uruchomienia innego zadania.

Następujące klucze są dostępne dla zadania uruchomienia. Aby uzyskać odpowiednią definicję obiektu interfejsu API REST, zobacz run_job_task.

Przykłady

Poniższy przykład zawiera zadanie uruchamiania zadania w drugim zadaniu, które uruchamia pierwsze zadanie.

W tym przykładzie użyto podstawienia, aby pobrać identyfikator zadania do uruchomienia. Aby uzyskać identyfikator zadania z interfejsu użytkownika, otwórz zadanie w obszarze roboczym i skopiuj identyfikator z wartości Identyfikator zadania na karcie Szczegóły zadania na stronie ustawień zadania.

resources:
  jobs:
    my-first-job:
      name: my-first-job
      tasks:
        - task_key: my-first-job-task
          new_cluster:
            spark_version: '13.3.x-scala2.12'
            node_type_id: 'i3.xlarge'
            num_workers: 2
          notebook_task:
            notebook_path: ./src/test.py
    my_second_job:
      name: my-second-job
      tasks:
        - task_key: my-second-job-task
          run_job_task:
            job_id: ${resources.jobs.my-first-job.id}

Zadanie SQL

To zadanie służy do uruchamiania pliku SQL, zapytania lub alertu.

Następujące klucze są dostępne dla zadania SQL. Aby uzyskać odpowiednią definicję obiektu interfejsu API REST, zobacz sql_task.

Przykłady

Poniższy przykład dodaje zadanie pliku SQL do zadania. To zadanie dotyczące pliku SQL korzysta z określonego magazynu SQL Warehouse do uruchomienia tego pliku SQL.

resources:
  jobs:
    my-sql-file-job:
      name: my-sql-file-job
      tasks:
        - task_key: my-sql-file-task
          sql_task:
            file:
              path: /Users/someone@example.com/hello-world.sql
              source: WORKSPACE
            warehouse_id: 1a111111a1111aa1

Poniższy przykład dodaje zadanie alertu SQL do zadania. To zadanie alertu SQL używa określonego magazynu SQL w celu odświeżenia określonego alertu SQL.

resources:
  jobs:
    my-sql-file-job:
      name: my-sql-alert-job
      tasks:
        - task_key: my-sql-alert-task
          sql_task:
            warehouse_id: 1a111111a1111aa1
            alert:
              alert_id: 11111111-1111-1111-1111-111111111111

Poniższy przykład dodaje zadanie zapytania SQL do zadania. To zadanie zapytania SQL używa określonego magazynu SQL do uruchomienia określonego zapytania SQL.

resources:
  jobs:
    my-sql-query-job:
      name: my-sql-query-job
      tasks:
        - task_key: my-sql-query-task
          sql_task:
            warehouse_id: 1a111111a1111aa1
            query:
              query_id: 11111111-1111-1111-1111-111111111111

Inne ustawienia zadania

Następujące ustawienia zadania umożliwiają konfigurowanie zachowań dla wszystkich zadań. Aby zapoznać się z odpowiednimi definicjami obiektów interfejsu API REST, zobacz zadania.