Data Management:Schedule a notebook with Airflow

Prerequisites

• You have created an Airflow instance.

• You have added an associated account.

• You have created a code repository.

Schedule a notebook with Airflow

1. Log in to DMS 5.0.

2. Go to the workspace page.

   DMS provides two ways to access the workspace.

   Method 1

   Move the pointer over the icon in the upper-left corner and choose All Features > Data+AI > Dify.

   Note

   If you use the DMS console in normal mode, choose Data+AI > Dify in the top navigation bar.

   Method 2

   In the left-side navigation pane, click the Digital Factory icon, and then click Workspace.

   Note

   If you are using the non-minimalist console mode, choose Digital Factory > Workspace from the top navigation bar.

3. In the WORKSPACE area of the (Resource Manager) page, click and select New Notebook File.

4. In the notebook file, add your code, such as print(1).

5. In the REPOS (code repository) section, write Python code to configure the scheduling parameters for the notebook file. The following is an example:

   from doctest import debug
   from airflow import DAG
   from airflow.decorators import task
   from airflow.models.param import Param
   from airflow.operators.bash import BashOperator
   from airflow.operators.empty import EmptyOperator
   from airflow.providers.alibaba_dms.cloud.operators.dms_notebook import DMSNotebookOperator
   with DAG(
       "dms_notebook_sy_hz_name",
       params={
       },
   ) as dag:
       notebook_operator = DMSNotebookOperator(
           task_id='dms_notebook_sy_hz_name',
           profile_name='test',
           profile={},
           cluster_type='spark',
           cluster_name='spark_cluster_855298',
           spec='4C16G',
           runtime_name='Spark3.5_Scala2.12_Python3.9_General:1.0.9',
           file_path='/Workspace/code/default/test.ipynb',
           run_params={'a':10},    
           polling_interval=5,
           debug=True,
           dag=dag
       )
       run_this_last = EmptyOperator(
           task_id="run_this_last22",
           dag=dag,
       )
       notebook_operator >> run_this_last
   if __name__ == "__main__":
       dag.test(
           run_conf={}
       )

   The following table describes key parameters. For parameters not listed, you can use the default values.

   Parameter	Type	Required	Description
   task_id	string	Yes	A unique identifier for the task.
   profile_name	string	Yes	The profile name. Click the (Configuration Management) icon on the right border to configure a new profile.
   cluster_type	string	Yes	The cluster type for the notebook session instance. Supported types are CPU and Spark. DMS creates CPU clusters by default; you must create Spark clusters manually. For more information, see Create a Spark cluster.
   cluster_name	string	Yes	The cluster name.
   spec	string	Yes	The cluster specifications. Currently, only the default specification, 4C16G, is supported.
   runtime_name	string	Yes	The runtime environment. For Spark, only the Spark3.5_Scala2.12_Python3.9_General:1.0.9 and Spark3.3_Scala2.12_Python3.9_General:1.0.9 runtime environments are supported.
   file_path	string	Yes	The path to the notebook file. The path format is /Workspace/code/default. Example: /Workspace/code/default/test.ipynb.
   run_params	dict	No	Parameters used to replace variables in the notebook file at runtime.
   timeout	int	No	The maximum execution time for a notebook cell, in seconds. If a cell exceeds this timeout, the notebook's scheduling task fails.
   polling_interval	int	No	The interval, in seconds, at which to refresh execution results. Default: 10.

6. Bind a publishing environment to the code repository.

   Click the name of the code repository. On the environment configuration page, bind the environment to the workflow instance to enable publishing. A workflow instance can be bound to only one type of environment.

7. Hover over the name of the target repository, click the icon, select Publish and the Publishing Environment (the environment bound to the repository), and then click OK.

   Note

   The publishing operation has a 10-second delay.

8. Run the notebook file.

   1. In the workspace, click the workflow icon on the left, and then click the name of the target Airflow instance to open the Airflow UI.

   2. On the Code page, verify that the published code has been synchronized.

   3. Once the code is synchronized, click the Run icon in the upper-right corner.

   4. Click the Graph tab, and then find and click the corresponding task.

   5. Click the Logs tab to view the logs for the task.

      While the task is running, you can view the execution progress of the notebook file.

View notebook progress

• View progress on the DAG page

  On the Logs tab, view the execution progress of the current task. For example, a progress of 2/15 indicates that the second of 15 total cells is running.

  Note

  When the message notebook run success appears, the task is complete.

• View progress on the notebook page

  On the Logs page, find the url for the notebook page in the logs. Click this link to open the notebook file and view its execution status. You can click the refresh button in the upper-right corner to view real-time progress.

  When the execution results appear, the notebook task is complete.

  When the notebook file finishes executing, its last execution time and a Success status are displayed at the top of the page. The output and time consumed for each code cell are displayed below that cell.