Tutorial 2: Crie e gerencie tarefas e gráficos de tarefa (DAGs)¶

Introdução¶

Neste tutorial, você cria e usa tarefas Snowflake para gerenciar alguns procedimentos armazenados básicos. Você também cria um gráfico de tarefa – também chamado de gráfico acíclico dirigido (DAG) – para orquestrar tarefas com uma API de gráfico de tarefa de nível superior.

Configuração de objetos Snowflake¶

Configure os procedimentos armazenados que suas tarefas invocarão e o estágio que conterá os procedimentos armazenados. É possível usar seu objeto root do Snowflake Python APIs para criar um estágio no banco de dados PYTHON_API_DB e no esquema PYTHON_API_SCHEMA criado anteriormente.

1. Para criar um estágio chamado TASKS_STAGE, na próxima célula do notebook, execute o seguinte código:

   stages = root.databases[database.name].schemas[schema.name].stages
   stages.create(Stage(name="TASKS_STAGE"))
   

   Copy

   Este estágio manterá os procedimentos armazenados e quaisquer dependências que esses procedimentos precisem.

2. Para criar duas funções básicas do Python, que as tarefas executarão como procedimentos armazenados, na próxima célula, execute o seguinte código:

   def trunc(session: Session, from_table: str, to_table: str, count: int) -> str:
     (
       session
       .table(from_table)
       .limit(count)
       .write.save_as_table(to_table)
     )
     return "Truncated table successfully created!"
   
   def filter_by_shipmode(session: Session, mode: str) -> str:
     (
       session
       .table("snowflake_sample_data.tpch_sf100.lineitem")
       .filter(col("L_SHIPMODE") == mode)
       .limit(10)
       .write.save_as_table("filter_table")
     )
     return "Filter table successfully created!"
   

   Copy

   Essas funções fazem o seguinte:

   • trunc(): Cria uma versão truncada de uma tabela de entrada.

   • filter_by_shipmode(): Filtra a tabela SNOWFLAKE_SAMPLE_DATA.TPCH_SF100.LINEITEM por modo de envio, limita os resultados a 10 linhas e grava os resultados em uma nova tabela.

     Nota

     Esta função consulta os dados de amostra TPC-H no banco de dados SNOWFLAKE_SAMPLE_DATA. O Snowflake cria o banco de dados de amostra em novas contas por padrão. Se o banco de dados não foi criado em sua conta, consulte Uso do banco de dados de amostra.

   As funções são intencionalmente básicas e destinadas a fins de demonstração.

Criação e gerenciamento de tarefas¶

Defina, crie e gerencie duas tarefas que executarão suas funções Python criadas anteriormente como procedimentos armazenados.

1. Para definir as duas tarefas, task1 e task2, na próxima célula do notebook, execute o seguinte código:

   tasks_stage = f"{database.name}.{schema.name}.TASKS_STAGE"
   
   task1 = Task(
       name="task_python_api_trunc",
       definition=StoredProcedureCall(
         func=trunc,
         stage_location=f"@{tasks_stage}",
         packages=["snowflake-snowpark-python"],
       ),
       warehouse="COMPUTE_WH",
       schedule=timedelta(minutes=1)
   )
   
   task2 = Task(
       name="task_python_api_filter",
       definition=StoredProcedureCall(
         func=filter_by_shipmode,
         stage_location=f"@{tasks_stage}",
         packages=["snowflake-snowpark-python"],
       ),
       warehouse="COMPUTE_WH"
   )
   

   Copy

   Neste código, você especifica os seguintes parâmetros de tarefa:

   • Para cada tarefa, uma definição representada por um objeto StoredProcedureCall que inclui os seguintes atributos:

     • A função chamável a ser executada

     • O local do estágio onde o conteúdo da função Python e suas dependências são carregados

     • Dependências do pacote do procedimento armazenado

   • Um warehouse para executar o procedimento armazenado (necessário ao criar uma tarefa com um objeto StoredProcedureCall). Este tutorial usa o warehouse COMPUTE_WH incluído em sua conta de avaliação.

   • Um cronograma de execução para a tarefa raiz, task1. O cronograma especifica o intervalo no qual a tarefa deve ser executada periodicamente.

   Para obter mais informações sobre os procedimentos armazenados, consulte Como escrever procedimentos armazenados em Python.

2. Para criar as duas tarefas, recuperar um objeto TaskCollection (tasks) do esquema do banco de dados e chamar .create() na coleção de tarefas:

   # create the task in the Snowflake database
   tasks = schema.tasks
   trunc_task = tasks.create(task1, mode=CreateMode.or_replace)
   
   task2.predecessors = [trunc_task.name]
   filter_task = tasks.create(task2, mode=CreateMode.or_replace)
   

   Copy

   Neste exemplo de código, você também vincula as tarefas configurando task1 como predecessora para task2, o que cria um gráfico de tarefa mínimo.

3. Para confirmar que as duas tarefas agora existem, na próxima célula, execute o seguinte código:

   taskiter = tasks.iter()
   for t in taskiter:
       print(t.name)
   

   Copy

4. Quando você cria tarefas, elas são suspensas por padrão.

   Para iniciar uma tarefa, chame .resume() no objeto de recurso da tarefa:

   trunc_task.resume()
   

   Copy

5. Para confirmar que a tarefa trunc_task foi iniciada, na próxima célula, execute o seguinte código:

   taskiter = tasks.iter()
   for t in taskiter:
       print("Name: ", t.name, "| State: ", t.state)
   

   Copy

   A saída deve ser semelhante a esta:

   Name:  TASK_PYTHON_API_FILTER | State:  suspended
   Name:  TASK_PYTHON_API_TRUNC | State:  started
   

   É possível repetir esta etapa sempre que quiser confirmar o status das tarefas.

6. Para limpar os recursos da tarefa, primeiro suspenda a tarefa antes de descartá-la.

   Na próxima célula, execute o seguinte código:

   trunc_task.suspend()
   

   Copy

7. Para confirmar que a tarefa está suspensa, repita a etapa 5.

8. Opcional: Para descartar ambas as tarefas, na próxima célula, execute o seguinte código:

   trunc_task.drop()
   filter_task.drop()
   

   Copy

Crie e gerencie um gráfico de tarefa¶

Quando você coordena a execução de um grande número de tarefas, pode ser difícil gerenciar cada tarefa individualmente. O Snowflake Python APIs fornece funcionalidade para orquestrar tarefas com uma API de gráfico de tarefa de nível superior.

Um gráfico de tarefa, também chamado de gráfico acíclico direcionado (DAG), é uma série de tarefas compostas de uma tarefa raiz e tarefas filho, organizadas por suas dependências. Para obter mais informações, consulte Gerenciamento de dependências de tarefa com gráficos de tarefa.

1. Para criar e implementar um gráfico de tarefa, execute o seguinte código:

   dag_name = "python_api_dag"
   dag = DAG(name=dag_name, schedule=timedelta(days=1))
   with dag:
       dag_task1 = DAGTask(
           name="task_python_api_trunc",
           definition=StoredProcedureCall(
               func=trunc,
               stage_location=f"@{tasks_stage}",
               packages=["snowflake-snowpark-python"]),
           warehouse="COMPUTE_WH",
       )
       dag_task2 = DAGTask(
           name="task_python_api_filter",
           definition=StoredProcedureCall(
               func=filter_by_shipmode,
               stage_location=f"@{tasks_stage}",
               packages=["snowflake-snowpark-python"]),
           warehouse="COMPUTE_WH",
       )
       dag_task1 >> dag_task2
   dag_op = DAGOperation(schema)
   dag_op.deploy(dag, mode=CreateMode.or_replace)
   

   Copy

   Neste código, você faz o seguinte:

   • Crie um objeto de gráfico de tarefa chamando o construtor DAG e especificando um nome e uma cronograma.

   • Defina o gráfico de tarefa – tarefas específicas usando o construtor DAGTask. Observe que o construtor aceita os mesmos argumentos que você especificou para a classe StoredProcedureCall em uma etapa anterior.

   • Especifique dag_task1 como a tarefa raiz e predecessora para dag_task2 com sintaxe mais conveniente.

   • Implemente o gráfico de tarefa no esquema PYTHON_API_SCHEMA do banco de dados PYTHON_API_DB.

2. Para confirmar a criação do gráfico de tarefa, na próxima célula, execute o seguinte código:

   taskiter = tasks.iter()
   for t in taskiter:
       print("Name: ", t.name, "| State: ", t.state)
   

   Copy

   É possível repetir esta etapa sempre que quiser confirmar o status das tarefas.

3. Para iniciar o gráfico de tarefa iniciando a tarefa raiz, na próxima célula, execute o seguinte código:

   dag_op.run(dag)
   

   Copy

4. Para confirmar que a tarefa PYTHON_API_DAG$TASK_PYTHON_API_TRUNC foi iniciada, repita a etapa 2.

   Nota

   A chamada de função invocada pelo gráfico de tarefa não será bem-sucedida porque você não a está chamando com nenhum dos argumentos necessários. O objetivo desta etapa é apenas demonstrar como iniciar programaticamente o gráfico de tarefa.

5. Para soltar o gráfico de tarefa, na próxima célula, execute o seguinte código:

   dag_op.drop(dag)
   

   Copy

6. Limpe o objeto de banco de dados que você criou nestes tutoriais:

   database.drop()
   

   Copy

Qual é o próximo passo?¶

Parabéns! Neste tutorial, você aprendeu como criar e gerenciar tarefas e gráficos de tarefa usando o Snowflake Python APIs.