云服务器内容精选

  • 使用案例 案例场景 因数据规整要求,需要周期性地将多组 DLI 源数据表数据导入到对应的DLI目的表,如表1所示。 表1 需要导入的列表情况 源数据表名 目的表名 a_new a b_2 b c_3 c d_1 d c_5 e b_1 f 如果通过SQL节点分别执行导入脚本,需要开发大量脚本和节点,导致重复性工作。在这种情况下,我们可以使用For Each节点进行循环作业,节省开发工作量。 配置方法 准备源表和目的表。为了便于后续作业运行验证,需要先创建DLI源数据表和目的表,并给源数据表插入数据。 创建DLI表。您可以在 DataArts Studio 数据开发中,新建DLI SQL脚本执行以下SQL命令,也可以在 数据湖探索 (DLI)服务控制台中的SQL编辑器中执行以下SQL命令: /* 创建数据表 */ CREATE TABLE a_new (name STRING, score INT) STORED AS PARQUET; CREATE TABLE b_2 (name STRING, score INT) STORED AS PARQUET; CREATE TABLE c_3 (name STRING, score INT) STORED AS PARQUET; CREATE TABLE d_1 (name STRING, score INT) STORED AS PARQUET; CREATE TABLE c_5 (name STRING, score INT) STORED AS PARQUET; CREATE TABLE b_1 (name STRING, score INT) STORED AS PARQUET; CREATE TABLE a (name STRING, score INT) STORED AS PARQUET; CREATE TABLE b (name STRING, score INT) STORED AS PARQUET; CREATE TABLE c (name STRING, score INT) STORED AS PARQUET; CREATE TABLE d (name STRING, score INT) STORED AS PARQUET; CREATE TABLE e (name STRING, score INT) STORED AS PARQUET; CREATE TABLE f (name STRING, score INT) STORED AS PARQUET; 给源数据表插入数据。您可以在DataArts Studio数据开发模块中,新建DLI SQL脚本执行以下SQL命令,也可以在 数据湖 探索(DLI)服务控制台中的SQL编辑器中执行以下SQL命令: /* 源数据表插入数据 */ INSERT INTO a_new VALUES ('ZHAO','90'),('QIAN','88'),('SUN','93'); INSERT INTO b_2 VALUES ('LI','94'),('ZHOU','85'); INSERT INTO c_3 VALUES ('WU','79'); INSERT INTO d_1 VALUES ('ZHENG','87'),('WANG','97'); INSERT INTO c_5 VALUES ('FENG','83'); INSERT INTO b_1 VALUES ('CEHN','99'); 准备数据集数据。您可以通过以下方式之一获取数据集: 您可以将表1数据导入到DLI表中,然后将SQL脚本读取的结果作为数据集。 您可以将表1数据保存在OBS的 CS V文件中,然后通过DLI SQL或DWS SQL创建OBS外表关联这个CSV文件,然后将OBS外表查询的结果作为数据集。DLI创建外表请参见OBS输入流,DWS创建外表请参见创建外表。 您可以将表1数据保存在HDFS的CSV文件中,然后通过HIVE SQL创建Hive外表关联这个CSV文件,然后将HIVE外表查询的结果作为数据集。 MRS 创建外表请参见创建表。 本例以方式1进行说明,将表1中的数据导入到DLI表(Table_List)中。您可以在DataArts Studio数据开发模块中,新建DLI SQL脚本执行以下SQL命令导入数据,也可以在数据湖探索(DLI)服务控制台中的SQL编辑器中执行以下SQL命令: /* 创建数据表TABLE_LIST,然后插入表1数据,最后查看生成的表数据 */ CREATE TABLE Table_List (Source STRING, Destination STRING) STORED AS PARQUET; INSERT INTO Table_List VALUES ('a_new','a'),('b_2','b'),('c_3','c'),('d_1','d'),('c_5','e'),('b_1','f'); SELECT * FROM Table_List; 生成的Table_List表数据如下: 图2 Table_List表数据 创建要循环运行的子作业ForeachDemo。在本次操作中,定义循环执行的是一个包含了DLI SQL节点的任务。 进入DataArts Studio数据开发模块选择“作业开发”页面,新建作业ForeachDemo,然后选择DLI SQL节点,编排图3所示的作业。 DLI SQL的语句中把要替换的变量配成${}这种参数的形式。在下面的SQL语句中,所做的操作是把${Source}表中的数据全部导入${Destination}中,${fromTable}、${toTable} 就是要替换的变量参数。SQL语句为: INSERT INTO ${Destination} select * from ${Source}; 此处不能使用EL表达式#{Job.getParam("job_param_name")} ,因为此表达式只能直接获取当前作业里配置的参数的value,并不能获取到父作业传递过来的参数值,也不能获取到工作空间里面配置的全局变量,作用域仅为本作业。 而表达式${job_param_name},既可以获取到父作业传递过来的参数值,也可以获取到全局配置的变量。 图3 循环执行子作业 配置完成SQL语句后,在子作业中配置作业参数。此处仅需要配置参数名,用于主作业ForeachDemo_master中的For Each节点识别子作业参数;参数值无需填写。 图4 配置子作业参数 配置完成后保存作业。 创建For Each节点所在的主作业ForeachDemo_master。 进入DataArts Studio数据开发模块选择“作业开发”页面,新建数据开发主作业ForeachDemo_master。选择DLI SQL节点和For Each节点,选中连线图标并拖动,编排图5所示的作业。 图5 编排作业 配置DLI SQL节点属性,此处配置为SQL语句,语句内容如下所示。DLI SQL节点负责读取DLI表Table_List中的内容作为数据集。 SELECT * FROM Table_List; 图6 DLI SQL节点配置 配置For Each节点属性。 子作业:子作业选择步骤2已经开发完成的子作业“ForeachDemo”。 数据集:数据集就是DLI SQL节点的Select语句的执行结果。使用EL表达式#{Job.getNodeOutput('preDLI')},其中preDLI为前一个节点的名称。 子作业参数:用于将数据集中的数据传递到子作业以供使用。Source对应的是数据集Table_List表的第一列,Destination是第二列,所以配置的EL表达式分别为#{Loop.current[0]}、#{Loop.current[1]}。 图7 配置For Each节点 配置完成后保存作业。 测试运行主作业。 单击主作业画布上方的“测试运行”按钮,测试作业运行情况。主作业运行后,会通过For Each节点自动调用运行子作业。 单击左侧导航栏中的“实例监控”,进入实例监控中查看作业运行情况。等待作业运行成功后,就能查看For Each节点生成的子作业实例,由于数据集中有6行数据,所以这里就对应产生了6个子作业实例。 图8 查看作业实例 查看对应的6个DLI目的表中是否已被插入预期的数据。您可以在DataArts Studio数据开发模块中,新建DLI SQL脚本执行以下SQL命令导入数据,也可以在数据湖探索(DLI)服务控制台中的SQL编辑器中执行以下SQL命令: /* 查看表a数据,其他表数据请修改命令后运行 */ SELECT * FROM a; 将查询到的表数据与给源数据表插入数据步骤中的数据进行对比,可以发现数据插入符合预期。 图9 目的表数据
  • 适用场景 当您进行作业开发时,如果某些任务的参数有差异、但处理逻辑全部一致,在这种情况下您可以通过For Each节点避免重复开发作业。 For Each节点可指定一个子作业循环执行,并通过数据集对子作业中的参数进行循环替换。关键参数如下: 子作业:选择需要循环执行的作业。 数据集:即不同子任务的参数值的集合。可以是给定的数据集,如“[['1'],['3'],['2']]”;也可以是EL表达式如“#{Job.getNodeOutput('preNodeName')}”,即前一个节点的输出值。 子作业参数:参数名即子作业中定义的变量;参数值一般配置为数据集中的某组数据,每次运行中会将参数值传递到子作业以供使用。例如参数值填写为:#{Loop.current[0]},即将数据集中每行数据的第一个数值遍历传递给子作业。 For Each节点举例如图1所示 。从图中可以看出,子作业“foreach”中的参数名为“result”,参数值为一维数组数据集“[['1'],['3'],['2']] ”的遍历(即第一次循环为1,第二次循环为3,第三次循环为2)。 图1 for each节点
  • For Each节点与EL表达式 要想使用好For Each节点,您必须对EL表达式有所了解。EL表达式用法请参考EL表达式。 下面为您展示For Each节点常用的一些EL表达式。 #{Loop.dataArray} :For循环节点输入的数据集,是一个二维数组。 #{Loop.current}:由于For循环节点在处理数据集的时候,是一行一行进行处理的,那Loop.current就表示当前处理到的某行数据,Loop.current是一个一维数组,一般定义格式为#{Loop.current[0]}、#{Loop.current[1]}或其他,0表示遍历到当前行的第一个值。 #{Loop.offset}:For循环节点在处理数据集时当前的偏移量,从0开始。 #{Job.getNodeOutput('preNodeName')}:获取前面节点的输出。
  • 方案说明 DataArts Studio数据开发模块支持以事件触发的方式运行作业,因此通过DIS或者MRS Kafka作为作业依赖纽带,可以跨空间实现作业调度。 如下图,工作空间A中的job1运行完成后,可以使用DIS Client或Kafka Client发送消息触发中继作业job_agent;job_agent配置事件触发调度,根据DIS Client或Kafka Client发送的消息触发运行后,判断消息是否符合预期,符合则触发job2作业运行,否则不再触发job2运行。 图1 调度方案
  • 配置方法(Kafka Client) 登录DataArts Studio控制台,找到所需要的DataArts Studio实例,单击实例卡片上的“进入控制台”,进入概览页面。 单击第一个工作空间A的“数据开发”,系统跳转至数据开发页面,新建数据开发作业job1。分别选择Dummy节点和Kafka Client节点,选中连线图标并拖动,编排如图7所示的作业。 Dummy节点不执行任何操作,本例选择Dummy节点仅为演示操作,实际使用中您可以用其他作业节点替代。 Kafka Client节点用于发送消息。您需要选择Kafka连接和Topic名称,并将发送数据配置为EL表达式job1,#{DateUtil.getDay(Job.startTime)}。则当本作业执行完成后,将使用Kafka Client发送一条字符串消息:job1,作业执行日期。例如2月15日作业job1执行,实际的消息则为:job1,15。 作业调度等其他作业参数无需配置,保持默认即可。 图7 job1作业Kafka Client节点配置 在另一个工作空间B,新建数据开发作业job_agent。分别选择Dummy节点和Subjob节点,选中连线图标并拖动,编排图8所示的作业。 图8 job_agent作业调度配置 Dummy节点不执行任何操作,本例选择Dummy节点用于设置Dummy节点到Subjob节点之间连线的IF条件。 Subjob节点用于将需要后续执行的作业job2作为子作业引用执行。实际使用中您可以引用已有作业,也可以使用其他作业节点替代Subjob节点。 作业的调度方式设置为“事件驱动调度”,连接名称和Topic选择为工作空间B中的Kafka连接和Topic,需要与工作空间A中job1作业中Kafka Client节点所选择的Kafka连接和Topic相对应,用于通过Kafka消息触发作业运行。 IF判断条件设置,用于校验Kafka Client节点发送的消息是否符合预期,符合才会继续执行Subjob节点,否则跳过。 右键单击连线,选择“设置条件”,在弹出的“编辑参数表达式”文本框中输入IF判断条件,失败策略保持默认即可。IF判断条件为通过EL表达式语法填写三元表达式,当三元表达式结果为true的时候,才会执行连线后面的节点,否则后续节点将被跳过。 #{StringUtil.equals(StringUtil.split(Job.eventData,',')[1],'21')} 该IF判断条件表示,仅当从Kafka通道获取的消息逗号后的部分为“21”时,即每月21日时,才执行后续的作业节点。 如果您需要匹配多条消息记录,可以添加多个Dummy节点并分别添加到Subjob节点的IF条件,然后将数据开发组件配置项中的“多IF策略”设置为“逻辑或”即可。 图9 编辑参数表达式 测试运行作业job_agent,在工作空间A的作业job1未运行的情况下,前往实例监控中查看执行结果是否符合预期。 由于作业job1未运行即未发送消息,则job_agent作业中的Subjob节点被跳过,证明IF条件判断生效。 图10 Subjob节点被跳过 启动调度job_agent。然后测试运行工作空间A作业job1,待job1实例运行成功后,前往工作空间B实例监控中查看作业运行结果是否符合预期。 job_agent被触发运行。 如果当天日期和IF条件中的日期匹配,则job_agent作业中的Subjob节点成功运行、子作业job2也执行完成。否则Subjob节点被跳过。 图11 Subjob节点成功运行
  • 配置方法(DIS Client) 登录DataArts Studio控制台,找到所需要的DataArts Studio实例,单击实例卡片上的“进入控制台”,进入概览页面。 单击第一个工作空间A的“数据开发”,系统跳转至数据开发页面,新建数据开发作业job1。分别选择Dummy节点和DIS Client节点,选中连线图标并拖动,编排如图2所示的作业。 Dummy节点不执行任何操作,本例选择Dummy节点仅为演示操作,实际使用中您可以用其他作业节点替代。 DIS Client节点用于发送消息。您需要选择DIS所属Region和通道,并将发送数据配置为EL表达式job1,#{DateUtil.getDay(Job.startTime)}。则当本作业执行完成后,将使用DIS Client发送一条字符串消息:job1,作业执行日期。例如2月15日作业job1执行,实际的消息则为:job1,15。 作业调度等其他作业参数无需配置,保持默认即可。 图2 job1作业DIS Client节点配置 在另一个工作空间B,新建数据开发作业job_agent。分别选择Dummy节点和Subjob节点,选中连线图标并拖动,编排图3所示的作业。 图3 job_agent作业调度配置 Dummy节点不执行任何操作,本例选择Dummy节点用于设置Dummy节点到Subjob节点之间连线的IF条件。 Subjob节点用于将需要后续执行的作业job2作为子作业引用执行。实际使用中您可以引用已有作业,也可以使用其他作业节点替代Subjob节点。 作业的调度方式设置为“事件驱动调度”,DIS通道名称选择为工作空间A中job1作业中DIS Client节点所选择的通道,用于通过DIS消息触发作业运行。 IF判断条件设置,用于校验DIS Client节点发送的消息是否符合预期,符合才会继续执行Subjob节点,否则跳过。 右键单击连线,选择“设置条件”,在弹出的“编辑参数表达式”文本框中输入IF判断条件,失败策略保持默认即可。IF判断条件为通过EL表达式语法填写三元表达式,当三元表达式结果为true的时候,才会执行连线后面的节点,否则后续节点将被跳过。 #{StringUtil.equals(StringUtil.split(Job.eventData,',')[1],'21')} 该IF判断条件表示,仅当从DIS通道获取的消息逗号后的部分为“21”时,即每月21日时,才执行后续的作业节点。 如果您需要匹配多条消息记录,可以添加多个Dummy节点并分别添加到Subjob节点的IF条件,然后将数据开发组件配置项中的“多IF策略”设置为“逻辑或”即可。 图4 编辑参数表达式 测试运行作业job_agent,在工作空间A的作业job1未运行的情况下,前往实例监控中查看执行结果是否符合预期。 由于作业job1未运行即未发送消息,则job_agent作业中的Subjob节点被跳过,证明IF条件判断生效。 图5 Subjob节点被跳过 启动调度job_agent。然后测试运行工作空间A作业job1,待job1实例运行成功后,前往工作空间B实例监控中查看作业运行结果是否符合预期。 job_agent被触发运行。 如果当天日期和IF条件中的日期匹配,则job_agent作业中的Subjob节点成功运行、子作业job2也执行完成。否则Subjob节点被跳过。 图6 Subjob节点成功运行