本文为您介绍如何通过阿里云E-MapReduce（简称EMR）控制台，快速创建EMR Studio集群并开展交互式开发和工作流调度工作。

背景信息

如果您想了解更多Zeppelin、Jupyter和Airflow的信息，请参见以下内容：

前提条件

  • 已申请体验EMR Studio的资格。
  • 已创建EMR Studio集群，详情请参见创建EMR Studio集群。
    说明 如果您是第一次创建EMR Studio集群，在创建EMR Studio集群时，控制台会弹出系统角色授权窗口，请您使用主账号对系统角色AliyunECSInstanceForEMRStudioRole进行授权。角色授权详情请参见角色授权。
  • 安全组规则已开启8000、8081和8443端口。

    添加安全组规则，详情请参见添加安全组规则。

  • 已添加用户，详情请参见添加用户。

使用限制

EMR Studio仅支持与同一VPC下的EMR计算集群进行关联。

操作流程

  1. 步骤一：进入数据开发工作台
  2. 步骤二：关联计算集群
  3. 步骤三：使用Zeppelin交互式开发作业
  4. 步骤四：编写Airflow Python脚本
  5. 步骤五：使用Airflow调度Notebook作业

步骤一：进入数据开发工作台

  1. 进入详情页面。
    1. 登录阿里云E-MapReduce控制台。
    2. 在顶部菜单栏处，根据实际情况选择地域和资源组。
    3. 单击上方的集群管理页签。
    4. 在集群管理页面，单击相应集群所在行的详情。
  2. 添加Linux用户。
    您可以通过用户管理功能添加用户，详情请参见添加用户。
  3. 在左侧导航栏中，选择访问链接与端口。
  4. 单击Studio Workspace UI所在行的链接。
    输入步骤2中添加的用户名和密码，即可正常访问Web UI页面。
    说明 请稍等几秒即可看到Studio Workspace UI的信息。

步骤二：关联计算集群

重要 EMR Studio仅支持与同一VPC下的EMR计算集群进行关联。

关联集群后，EMR Studio集群可以访问对应EMR集群上的资源并提交Job。

  1. 在关联集群页面，单击可关联集群页签。
    说明 仅显示同一个VPC下的EMR集群。支持关联Hadoop集群、Dataflow集群和DataScience集群三种集群类型。
  2. 选择待关联集群的集群类型。
  3. 单击待关联集群操作列的关联集群。
  4. 在关联集群对话框中，单击绑定。
    待已关联集群页签，显示关联的集群信息时，表示关联成功。
    说明 绑定集群过程大约需要1~2分钟，请您耐心等待。

步骤三：使用Zeppelin交互式开发作业

关联集群后，您可以在Zeppelin Notebook中进行交互式开发。EMR数据开发集群自带教程，本文以Airflow调度教程1为例介绍。

  1. 在左侧导航栏中，单击Zeppelin。
  2. 在Zeppelin页面，选择阿里云EMR数据开发教程 > Airflow教程 > Airflow调度教程1。
    页面展示如下图所示。Airflow1
    区域 描述
    ① Markdown语言（以%md开头），主要是为了添加一些说明文字。
    ② Shell脚本语言（以%sh开头），您可以运行Shell脚本，生成数据。
    ③ Hive（以%hive开头），您可以执行任意的Hive SQL语句。
    ④ 单击drop-down图标，可以切换关联的集群。
    ⑤ 单击run图标，运行当前段落。
    说明 部分解释器（Interpreter）在第一次使用的时候会去下载依赖，如果提示信息类似Interpreter Setting 'hive' is not ready，说明下载过程还没结束，请稍后重试。

步骤四：编写Airflow Python脚本

Airflow的调度需要手动编写Python脚本来构建DAG，EMR Studio自动将指定OSS路径内的Python脚本同步至Airflow DAGs，因此，您可以在编辑和上传完DAG脚本之后，进入数据开发工作台，在左侧导航栏中，单击Airflow，即可进入DAGs页面查看您创建的DAG。

EMR Studio自带调度教程，您可以在Zeppelin页面，选择阿里云EMR数据开发教程 > Airflow教程 > Airflow调度教程1查看。Airflow的基本用法，请参见Apache Airflow。Airflow dag
说明 EMR Studio自带用于调度Zeppelin Notebook的Operator（ZeppelinOperator），并提供Note和Paragraph两种级别调度方式。更多信息，请参见定期调度Zeppelin中的作业。
本文以Paragraph级别调度为例，为每个Paragraph构建一个Task，然后串联起来构成一个链式DAG。代码示例如下。
from airflow import DAG
from datetime import datetime, timedelta
from airflow.providers.apache.zeppelin.operators.zeppelin_operator import ZeppelinOperator

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2018, 1, 1),
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 0,
    'retry_delay': timedelta(minutes=5),
}


with DAG('zeppelin_example_paragraph_dag',
         max_active_runs=5,
         schedule_interval='0 0 * * *',
         default_args=default_args) as dag:

    execution_date = "{{ ds }}"

    raw_data_task = ZeppelinOperator(
        task_id='raw_data_task',
        conn_id='zeppelin_default',
        note_id='2FXADMCY9',
        paragraph_id='paragraph_1613703810025_1489985581',
        params= {'dt' : execution_date}
    )

    csv_data_task = ZeppelinOperator(
        task_id='csv_data_task',
        conn_id='zeppelin_default',
        note_id='2FXADMCY9',
        paragraph_id='paragraph_1613705325984_828524031',
        params= {'dt' : execution_date}
    )

    parquet_data_task = ZeppelinOperator(
        task_id='parquet_data_task',
        conn_id='zeppelin_default',
        note_id='2FXADMCY9',
        paragraph_id='paragraph_1613709950691_1420422787',
        params= {'dt' : execution_date}
    )

    query_task = ZeppelinOperator(
        task_id='query_task',
        conn_id='zeppelin_default',
        note_id='2FXADMCY9',
        paragraph_id='paragraph_1613703819960_454383326',
        params= {'dt' : execution_date}
    )

    raw_data_task >> csv_data_task >> parquet_data_task >> query_task

步骤五：使用Airflow调度Notebook作业

  • 启用DAG调度
    在DAGs页面，打开off开关，即可启用DAG调度。DAG
  • 查看调度任务的详细信息

    在DAGs页面，单击待查看调度任务的DAG，进入Tree View页面。

    在Tree View页面，您可以查看对应DAG运行调度任务的详细信息。DAG_info
  • Task的运行情况

    在Tree View页面，单击task图标，即可查看对应Task的运行情况。

    详细信息如下图所示。task_log