更多操作 创建数据集完成后，可根据需要执行如表2所示的操作。 表2 更多操作 操作 步骤 查看数据集详情 在“微调数据集”页面选择“我创建的”页签。 在数据集列表中单击数据集名称，在微调数据集详情页面查看数据概况、调度历史，并支持对数据集进行溯源。 修改数据集 在“微调数据集”页面选择“我创建的”页签。 在数据集列表勾选数据集并单击“操作”列的“修改”。 在“修改数据集”页面，仅支持修改数据集描述、修改标签名称。 删除数据集 单个删除数据集： 在“我的数据集”页面选择“我创建的”页签。 在数据集列表勾选单个数据集，然后选择“操作”列的“删除”。 单击“确认”。 批量删除数据集： 在“我的数据集”页面选择“我创建的”页签。 在数据集列表勾选多个数据集，再单击列表上方“批量删除”。 在“批量删除”对话框，单击“确认”。 说明： 被标注的数据集无法删除。 标注数据集 说明： 只有格式为“对话文本”的数据集才可进行标注。 调度类型为“一次性调度”的数据集才可进行标注。 需要先在对微调数据集进行数据标注中创建标注任务，才能在当前页面执行数据标注。 在数据集列表中，单击数据集记录前的。 单击版本列表操作列的“标注”，参照对微调数据集进行数据标注进行数据标注。

前提条件 通过OBS（ 对象存储服务 ）接入数据时，操作账号需获得OBS只读权限，具体操作请参见对其他账号授予桶的读写权限。 需具备充足的知识库容量包资源（包含OBS存储配额和向量库存储配额，两者比例为5:1），每个租户默认具备5G的OBS存储配额，默认配额耗完后，请参考购买AppStage购买知识库容量包。 需要具备AI原生应用引擎管理员或开发者权限，权限申请操作请参见AppStage组织成员申请权限。

路径和表名的时间宏变量 如图1所示，如果将： 源端的“表名”配置为“ CDM _/${dateformat(yyyy-MM-dd)}”。 目的端的“写入目录”配置为“/opt/ttxx/${timestamp()}”。 经过宏定义转换，这个作业表示：将Oracle数据库的“SQOOP.CDM_20171016”表中数据，迁移到HDFS的“/opt/ttxx/1508115701746”目录中。 图1 源表名和写入目录配置为时间宏变量 目前也支持一个表名或路径名中有多个宏定义变量，例如“/opt/ttxx/${dateformat(yyyy-MM-dd)}/${timestamp()}”，经过转换后为“/opt/ttxx/2017-10-16/1508115701746”。

timestamp timestamp支持两种形式的参数： timestamp() 返回当前时间的时间戳，即从1970年到现在的毫秒数，如1508078516286。 timestamp(dateOffset, dateType) 返回经过时间偏移后的时间戳，“dateOffset”和“dateType”表示日期的偏移量以及偏移量的类型。 例如当前日期为“2017-10-16 09:00:00”，则“timestamp(-10, MINUTE)”返回当前时间点10分钟前的时间戳，即“1508115000000”。

时间变量宏定义具体展示 假设当前时间为“2017-10-16 09:00:00”，时间变量宏定义具体如表1所示。 表中示例实际使用是必须嵌在''中使用，比如需要以yyyy-MM-dd格式返回当前时间时，参数为'${dateformat(yyyy-MM-dd)}'。 表1 时间变量宏定义具体展示 宏变量 含义 实际显示效果 ${dateformat(yyyy-MM-dd)} 以yyyy-MM-dd格式返回当前时间。 2017-10-16 ${dateformat(yyyy/MM/dd)} 以yyyy/MM/dd格式返回当前时间。 2017/10/16 ${dateformat(yyyy_MM_dd HH:mm:ss)} 以yyyy_MM_dd HH:mm:ss格式返回当前时间。 2017_10_16 09:00:00 ${dateformat(yyyy-MM-dd HH:mm:ss, -1, DAY)} 以yyyy-MM-dd HH:mm:ss格式返回时间，时间为当前时间的前一天。 2017-10-15 09:00:00 ${dateformat(yyyy-MM-dd, -1, DAY)} 00:00:00 以yyyy-MM-dd HH:mm:ss格式返回时间，时间为当前时间的前一天0点。 2017-10-15 00:00:00 ${dateformat(yyyy-MM-dd, -1, DAY)} 12:00:00 以yyyy-MM-dd HH:mm:ss格式返回时间，时间为当前时间的前一天12点。 2017-10-15 12:00:00 ${dateformat(yyyy-MM-dd, -N, DAY)} 00:00:00 以yyyy-MM-dd HH:mm:ss格式返回时间，时间为当前时间的前N天的0点。 N为3时： 2017-10-13 00:00:00 ${dateformat(yyyy-MM-dd, -N, DAY)} 12:00:00 以yyyy-MM-dd HH:mm:ss格式返回时间，时间为当前时间的前N天的12点。 N为3时： 2017-10-13 12:00:00 ${timestamp()} 返回当前时间的时间戳，即1970年1月1日（00:00:00 GMT）到当前时间的毫秒数。 1508115600000 ${timestamp(-10, MINUTE)} 返回当前时间点10分钟前的时间戳。 1508115000000 ${timestamp(dateformat(yyyyMMdd))} 返回今天0点的时间戳。 1508083200000 ${timestamp(dateformat(yyyyMMdd,-1,DAY))} 返回昨天0点的时间戳。 1507996800000 ${timestamp(dateformat(yyyyMMddHH))} 返回当前整小时的时间戳。 1508115600000

Where子句中的时间宏变量 以SQOOP.CDM_20171016表为例，该表中存在表示时间的列DS，如图2所示。 图2 表数据 假设当前时间为“2017-10-16”，要导出前一天的数据（即DS=‘2017-10-15’），则可以在创建作业时配置“Where子句”为DS='${dateformat(yyyy-MM-dd,-1,DAY)}'，即可将符合DS=‘2017-10-15’条件的数据导出。

时间宏变量和定时任务配合完成增量同步 这里列举两个简单的使用场景： 数据库表中存在表示时间的列DS，类型为“varchar(30)”，插入的时间格式类似于“2017-xx-xx”。 定时任务中，重复周期为1天，每天的凌晨0点执行定时任务。配置“Where子句”为DS='${dateformat(yyyy-MM-dd,-1,DAY)}'，这样就可以在每天的凌晨0点导出前一天产生的所有数据。 数据库表中存在表示时间的列time，类型为“Number”，插入的时间格式为时间戳。 定时任务中，重复周期为1天，每天的凌晨0点执行定时任务。配置“Where子句”为time between ${timestamp(-1,DAY)} and ${timestamp()}，这样就可以在每天的凌晨0点导出前一天产生的所有数据。 其它的配置方式原理相同。

dateformat dateformat支持两种形式的参数： dateformat(format) format表示返回日期的格式，格式定义参考"java.text.SimpleDateFormat.java"中的定义。 例如当前日期为“2017-10-16 09:00:00”，则"yyyy-MM-dd HH:mm:ss"表示“2017-10-16 09:00:00”。 dateformat(format, dateOffset, dateType) format表示返回日期的格式。 dateOffset表示日期的偏移量。 dateType表示日期的偏移量的类型。 目前dateType支持以下几种类型：SECOND（秒），MINUTE（分钟），HOUR（小时），DAY（天），MONTH（月），YEAR（年）。 其中MONTH（月），YEAR（年）的偏移量类型存在特殊场景： 对于年、月来说，若进行偏移后实际没有该日期，则按照日历取该月最大的日期。 不支持在源端和目的端的“时间过滤”参数中的起始时间、终止时间使用年、月的偏移。 例如当前日期为"2023-03-01 09:00:00"，则： "dateformat(yyyy-MM-dd HH:mm:ss, -1, YEAR)"表示当前时间的前一年，也就是"2022-03-01 09:00:00"。 "dateformat(yyyy-MM-dd HH:mm:ss, -3, MONTH)"表示当前时间的前三月，也就是"2022-12-01 09:00:00"。 "dateformat(yyyy-MM-dd HH:mm:ss, -1, DAY)"表示当前时间的前一天，也就是"2023-02-28 09:00:00"。 "dateformat(yyyy-MM-dd HH:mm:ss, -1, HOUR)"表示当前时间的前一小时，也就是"2023-03-01 08:00:00"。 "dateformat(yyyy-MM-dd HH:mm:ss, -1, MINUTE)"表示当前时间的前一分钟，也就是"2023-03-01 08:59:00"。 "dateformat(yyyy-MM-dd HH:mm:ss, -1, SECOND)"表示当前时间的前一秒，也就是"2023-03-01 08:59:59"。

开发并调度作业 假设OBS中“movie”和“rating”表是每日更新的，我们希望每天更新Top10电影，那么这里可以使用DLF作业编排和作业调度功能。 在 DataArts Studio 控制台首页，选择对应工作空间的“数据开发”模块，进入数据开发页面。 创建一个DLF批处理作业，作业名称为“topmovie”。 图7 新建作业 图8 配置作业 在作业开发页面，拖动2个CDM Job节点、3个Dummy节点和2个DWS SQL节点到画布中，选中连线图标并拖动，编排图9所示的作业。 图9 连接和配置节点属性 关键节点说明： Begin（Dummy节点）：不执行任何操作，只作为起始点的标识。 movies_obs2dws（CDM Job节点）：在节点属性中，选择步骤2：数据集成入湖中的CDM集群，并关联CDM作业“movies_obs2dws”。 ratings_obs2dws（CDM Job节点）：在节点属性中，选择步骤2：数据集成入湖中的CDM集群，并关联CDM作业“ratings_obs2dws”。 Waiting（Dummy节点）：不执行任何操作，作为等待前侧节点执行结束的标识。 top_rating_movie（DWS SQL节点）：在节点属性中，关联创建DWS SQL脚本top_rating_movie中开发完成的DWS SQL脚本“top_rating_movie”。 top_active_movie（DWS SQL节点）：在节点属性中，关联创建DWS SQL脚本top_active_movie中开发完成的DWS SQL脚本“top_active_movie”。 Finish（Dummy节点）：不执行任何操作，只作为结束点的标识。 作业编排完成后，单击，测试运行作业。 如果作业运行正常，单击“调度配置”，配置作业的调度策略。 图10 调度配置 说明： 2022/02/09至2022/02/28，每天1点00分执行一次作业。 依赖属性：可以配置为依赖其他作业运行，本例不涉及，无需配置。 跨周期依赖：可以选择配置为依赖上一周期或者不依赖，此处配置为不依赖即可。 最后保存并提交版本（单击），执行调度作业（单击）。实现作业每天自动运行，Top10电影的结果自动保存到“top_active_movie”和“top_rating_movie”表。 您如果需要及时了解作业的执行结果是成功还是失败，可以通过数据开发的运维调度界面进行查看，如图11所示。 图11 查看作业执行情况

更多操作 一条数据标注完成后，可执行如下表1所示的操作。 表1 更多操作 操作 说明 删除标注 在“数据标注”页面的标注数据集列表中，单击“操作”列“删除”。 发布标注 在“数据标注”页面的标注数据集列表中，单击“操作”列“发布”。 在“发布”对话框，有两种发布方式： 选择发布后“更新原始数据集”，单击“确定”，覆盖原数据集信息（数据集名称不变）。 选择发布后“创建新数据集”，设置新数据集名称，然后单击“确定”。

标注数据集 在“数据标注”页面的标注数据集列表中，单击“操作”列“标注”。 在“标注信息”页面，在“数据集文件列表”下拉列表中选择文件。 单击“全部信息”页签下的“创建对话”顺次生成一条不完整信息（对话样式），用户根据实际需要填写对话的instruction（指令）、input（输入/提问）、output（输出/回答），完成一条数据标注。 对于单条标注，还可执行以下操作： 单击标注右侧“添加回答”可继续添加多条output。 单击标注右侧“删除”，可删除标注。 对于标注中的output，还可执行以下操作： 单击output所在行右侧的“自动生成”，由平台内置的模型自动生成当前行的output信息。 单击output所在行右侧的“重新生成”，由平台内置的模型重新生成当前行的output信息。 单击output所在行右侧的“删除”，可删除当前行的output信息。

示例代码 示例一：根据数据类型创建图像数据集 from modelarts.session import Session from modelarts.dataset import Dataset session = Session() dataset_name = "dataset-image" # 数据集名称 data_type = "IMAGE" # 数据集类型，图像类型数据集 data_sources = dict() # 数据集数据来源 data_sources["type"] = 0 # 数据来源类型，0表示OBS data_sources["path"] = "/obs-gaia-test/data/image/image-classification/" # 数据在OBS中的路径 work_path = dict() # 数据集输出位置，用于存放输出的标注信息等文件 work_path['type'] = 0 # 数据集工作目录的类型，0表示OBS work_path['path'] = "/obs-gaia-test/data/output/work_path/" # 数据集工作目录在OBS中的路径 create_dataset_resp = Dataset.create_dataset(session, dataset_name=dataset_name, data_type=data_type, data_sources=data_sources, work_path=work_path) 示例二：根据数据类型创建图像数据集（导入标注信息） from modelarts.session import Session from modelarts.dataset import Dataset session = Session() dataset_name = "dataset-image-with-annotations" data_type = "IMAGE" data_sources = dict() data_sources["type"] = 0 data_sources["path"] = "/obs-gaia-test/data/image/image-classification/" annotation_config = dict() # 源数据的标注格式 annotation_config['scene'] = "image_classification" # 数据标注场景为图像分类标注 annotation_config['format_name'] = "ModelArts image classification 1.0" # 标注格式为ModelArts image classification 1.0 data_sources['annotation_config'] = annotation_config work_path = dict() work_path['type'] = 0 work_path['path'] = "/obs-gaia-test/data/output/work_path/" create_dataset_resp = Dataset.create_dataset(session, dataset_name=dataset_name, data_type=data_type, data_sources=data_sources, work_path=work_path) 示例三：根据数据类型创建表格数据集 from modelarts.session import Session from modelarts.dataset import Dataset session = Session() dataset_name = "dataset-table" data_type = "TABLE" data_sources = dict() data_sources["type"] = 0 data_sources["path"] = "/obs-gaia-test/data/table/table0/" data_sources['with_column_header'] = True work_path = dict() work_path['type'] = 0 work_path['path'] = "/obs-gaia-test/data/output/work_path/" # 表格类型的数据集需要指定表格数据的schema信息 schema0 = dict() schema0['schema_id'] = 0 schema0['name'] = "name" schema0['type'] = "STRING" schema1 = dict() schema1['schema_id'] = 1 schema1['name'] = "age" schema1['type'] = "STRING" schema2 = dict() schema2['schema_id'] = 2 schema2['name'] = "label" schema2['type'] = "STRING" schemas = [] schemas.append(schema0) schemas.append(schema1) schemas.append(schema2) create_dataset_resp = Dataset.create_dataset(session, dataset_name=dataset_name, data_type=data_type, data_sources=data_sources, work_path=work_path, schema=schemas) 示例四：根据标注类型创建图像分类数据集 from modelarts.session import Session from modelarts.dataset import Dataset session = Session() dataset_name = "dataset-image-classification" dataset_type = 0 # 数据集的标注类型，0表示图像分类标注类型 data_sources = dict() data_sources["path"] = "/obs-gaia-test/data/image/image-classification/" data_sources["type"] = "0" work_path = dict() work_path['type'] = 0 work_path['path'] = "/obs-gaia-test/data/output/work_path/" create_dataset_resp = Dataset.create_dataset(session, dataset_name=dataset_name, dataset_type=dataset_type, data_sources=data_sources, work_path=work_path) 示例五：根据标注类型创建文本三元组数据集 dataset_name = "dataset-text-triplet" dataset_type = 102 # 数据集标注类型，102表示文本三元组标注类型 data_sources = dict() data_sources['type'] = 0 data_sources['path'] = "/obs-gaia-test/data/text/text-classification/" work_path = dict() work_path['type'] = 0 work_path['path'] = "/obs-gaia-test/data/output/work_path/" # 创建文本三元组标注类型的数据集，需要传入标签参数 label_entity1 = dict() # 标签对象 label_entity1['name'] = "疾病" # 标签名称 label_entity1['type'] = 101 # 标签类型，101表示实体类型标签 label_entity2 = dict() label_entity2['name'] = "疾病别称" label_entity2['type'] = 101 label_relation1 = dict() label_relation1['name'] = "又称为" label_relation1['type'] = 102 # 标签类型，102表示关系类型标签 property = dict() # 关系类型标签需要在标签属性中指定起始实体标签和终止实体标签 property['@modelarts:from_type'] = "疾病" # 起始实体标签 property['@modelarts:to_type'] = "疾病别称" # 终止实体标签 label_relation1['property'] = property labels = [] labels.append(label_entity1) labels.append(label_entity2) labels.append(label_relation1) create_dataset_resp = Dataset.create_dataset(session, dataset_name=dataset_name, dataset_type=dataset_type, data_sources=data_sources, work_path=work_path, labels=labels) 示例六：根据标注类型创建表格数据集 dataset_name = "dataset-table" dataset_type = 400 # 数据集标注类型，400表示表格数据集 data_sources = dict() data_sources['type'] = 0 data_sources['path'] = "/obs-gaia-test/data/table/table0/" data_sources['with_column_header'] = True # 用来指明表格数据中是否包含表头 work_path = dict() work_path['type'] = 0 work_path['path'] = "/obs-gaia-test/data/output/work_path/" # 表格数据集需要传入表格数据的表头参数 schema0 = dict() # 表格的表头 schema0['schema_id'] = 0 # 第一列表头 schema0['name'] = "name" # 表头名称，该列表头为name schema0['type'] = "STRING" # 表头数据类型，表示字符串 schema1 = dict() schema1['schema_id'] = 1 schema1['name'] = "age" schema1['type'] = "STRING" schema2 = dict() schema2['schema_id'] = 2 schema2['name'] = "label" schema2['type'] = "STRING" schemas = [] schemas.append(schema0) schemas.append(schema1) schemas.append(schema2) create_dataset_resp = Dataset.create_dataset(session, dataset_name=dataset_name, dataset_type=dataset_type, data_sources=data_sources, work_path=work_path, schema=schemas)

重视消息生产与消费的确认过程 消息生产 生产消息后，生产者需要根据ROMA Connect的返回信息确认消息是否发送成功，如果返回失败需要重新发送。 每次生产消息，生产者都需要等待消息发送API的应答信号，以确认消息是否成功发送。在消息传递过程中，如果发生异常，生产者没有接收到发送成功的信号，生产者自己决策是否需要重复发送消息。如果接收到发送成功的信号，则表明该消息已经被ROMA Connect可靠存储。 消息消费 消息消费时，消费者需要确认消息是否已被成功消费。 生产的消息被依次存储在ROMA Connect的存储介质中。消费时依次获取ROMA Connect中存储的消息。消费者获取消息后，进行消费并记录消费成功或失败的状态，并将消费状态提交到ROMA Connect，由ROMA Connect决定消费下一批消息或回滚重新消费消息。 在消费过程中，如果出现异常，没有提交消费确认，该批消息会在后续的消费请求中再次被获取。

消息生产与消费的幂等传递 ROMA Connect设计了一系列可靠性保障措施，确保消息不丢失。例如使用消息同步存储机制防止系统与服务器层面的异常重启或者掉电，使用消息确认（ACK）机制解决消息传输过程中遇到的异常。 考虑到网络异常等极端情况，用户除了做好消息生产与消费的确认，还需要配合ROMA Connect完成消息发送与消费的重复传输设计。 当无法确认消息是否已发送成功，生产者需要将消息重复发送给ROMA Connect。 当重复收到已处理过的消息，消费者需要告诉ROMA Connect消费成功且保证不重复处理。

消息可以批量生产和消费 为提高消息发送和消息消费效率，推荐使用批量消息发送和消费。通常，默认消息消费为批量消费，而消息发送尽可能采用批量发送，可以有效减少API调用次数。 如下面两张示意图对比所示，消息批量生产与消费，可以减少API调用次数，节约资源。 图1 消息批量生产与消费 批量发送消息时，单次不能超过10条消息，总大小不能超过512KB。 批量生产（发送）消息可以灵活使用，在消息并发多的时候，批量发送，并发少时，单条发送。这样能够在减少调用次数的同时保证消息发送的实时性。 图2 消息逐条生产与消费 此外，批量消费消息时，消费者应按照接收的顺序对消息进行处理、确认，当对某一条消息处理失败时，不再需要继续处理本批消息中的后续消息，直接对已正确处理的消息进行确认即可。