【华为云上云之路】网络AI模型开发（2）数据入湖、数据处理

本篇文章主要分享在华为云NAIE的数据服务中&＃xff0c;进行数据入湖、数据处理。

目录

数据入湖

数据处理

数据入湖

简介

  我们在华为云NAIE开通“数据资产管理服务”后&＃xff0c;把数据上传到数据湖OBS中。其中数据可以从本地文件、租户OBS、数据湖OBS、HIVE、DWS等方式上传。

流程

1&＃xff09;先准备数据&＃xff0c;以.csv为后缀的文件&＃xff1b;(本文中使用本地文件上传)

2&＃xff09;进入数据资产管理服务&＃xff0c;选择“治理工具”下面的“数据加载”&＃xff0c;把本地的数据上传数据到湖OBS中。

特点

1&＃xff09;数据输入端&＃xff1a;本地文件、租户OBS、数据湖OBS、HIVE、DWS等方式&＃xff1b;

2&＃xff09;数据生成端&＃xff1a;数据湖OBS。

数据资产管理服务 页面

进入服务&＃xff1b;

选择“治理工具”下面的“数据加载”&＃xff0c;点击“创建任务”

关于上传数据湖OBS的路径&＃xff0c;可以根据实际情况选择&＃xff0c;记得就好&＃xff0c;后面要用到的哈。

注意&＃xff1a;数据输入端&＃xff1a;本地文件、租户OBS、数据湖OBS、HIVE、DWS等方式&＃xff1b;根据实际情况选择哈。

如果想体验一下&＃xff0c;可以到网盘拿数据放到本地来上传&＃xff0c;链接&＃xff1a;https://pan.baidu.com/s/1KZiGw1_3EFnYBSKzp4VVlQ    提取码&＃xff1a;ifcj 
 

点击底下的连接测试&＃xff0c;当两端都连通后&＃xff0c;可以点击“提交”

查看状态&＃xff0c;在运行中&＃xff0c;等待一会变成运行成功&＃xff0c;就可以啦。

数据处理

简介

上传到数据湖OBS数据&＃xff0c;进行处理&＃xff1b;比如数据清洗&＃xff0c;数据过滤&＃xff0c;增删数据等&＃xff0c;也可以进行读写OBS的基本方法、参数的添加和使用。

特点

1&＃xff09;使用Python语言进行处理的

2&＃xff09;线上填写代码&＃xff1b;

3&＃xff09;支持不同的引擎框架&＃xff08;Horovod、MXNet、PyTorch、Ray、Spark_MLlib、TensorFlow、XGBoost-Sklearn&＃xff09;&＃xff0c;Python版本来运行&＃xff1b;

操作流程

来到“数据处理”&＃xff0c;点击“新增作业”

创建作业&＃xff1a;填写名称、选择作业位置等

点击“下一步”&＃xff0c;

main.py是官方的实例代码&＃xff0c;作用&＃xff1a;演示读写OBS的基本方法、参数的添加和使用方法

参考链接&＃xff1a;https://github.com/huaweicloud/ModelArts-Lab/blob/master/docs/moxing_api_doc/MoXing_API_File.md

"""
Sample Code, you can delete this file. Or clear this file content.
Execution engine suggests choosing TensorFlow/MXNet/PyTorch.
Python version 3.6 or higher.Functional description
1. Parsing running parameters. See function argument_init
2. Reading csv file from OBS. See function read_from_obs
3. Writing data to OBS. See function read_from_obsmoxing API document [https://github.com/huaweicloud/ModelArts-Lab/blob/master/docs/moxing_api_doc/MoXing_API_File.md]
"""
import argparse
import logging
import os
import pandas as pd# Adapt to ModelArts runtime environment
try:import moxing as moxmox.file.shift(&＃39;os&＃39;, &＃39;mox&＃39;)
except (ModuleNotFoundError, AttributeError):mox &＃61; Nonelogging.info(&＃39;Not exist moxing module. if running on modelarts engine. please select TensorFlow engine. Python Version is 3.6&＃39;)def argument_init():"""Parsing running parameters.:return: Running params"""logging.info(&＃39;Start to init running argument...&＃39;)parser &＃61; argparse.ArgumentParser(description&＃61;"pm data process")parser.add_argument(&＃39;--data_url&＃39;, type&＃61;str, default&＃61;&＃39;s3://bucket_name/folder/&＃39;, help&＃61;&＃39;System preset parameters.&＃39;)parser.add_argument(&＃39;--train_url&＃39;, type&＃61;str, default&＃61;&＃39;s3://bucket_name/folder/&＃39;, help&＃61;&＃39;System preset parameters.&＃39;)parser.add_argument(&＃39;--input_path&＃39;, type&＃61;str, default&＃61;None, help&＃61;&＃39;User-defined parameter example.&＃39;)parser.add_argument(&＃39;--output_path&＃39;, type&＃61;str, default&＃61;None, help&＃61;&＃39;User-defined parameter example.&＃39;)return parser.parse_known_args()def read_from_obs(input_path):"""Reading csv file from OBS:param input_path: OBS path. such as s3://bucket/folder/:return: data frame"""logging.info(&＃39;start to read file from OBS. input_path&＃61;%s&＃39; % input_path)if input_path is None or input_path.endswith(&＃39;s3://&＃39;):logging.error(&＃39;input_path is None or not endswith s3://&＃39;)returnif not os.path.exists(input_path):logging.warning(&＃39;input_path is not exists.&＃39;)return Nonedata_frame &＃61; Nonefor file_name in os.listdir(input_path):if not file_name.endswith(&＃39;.csv&＃39;):logging.info(&＃39;%s is not csv file.&＃39; % file_name)continuewith open(os.path.join(input_path, file_name), &＃39;rb&＃39;) as f:data_frame &＃61; pd.read_csv(f, encoding&＃61;&＃39;utf-8&＃39;)logging.info(&＃39;success read file. file_name&＃61;%s&＃39; % file_name)breakreturn data_framedef data_process(data_frame_before_process):"""User-defined data processing function:param data_frame: pandas data frame:return: Processed data frame"""logging.info(&＃39;start to process data...&＃39;)# Start of your own processing code.# Write your own processing code here.# End of your own processing code.logging.info(&＃39;end to process data...&＃39;)return data_frame_after_processdef write_to_obs(data_frame, output_path):"""Writing data to OBS.:param data_frame: pandas data frame:param output_path: OBS Path. such as s3://bucket/folder/:return: None"""logging.info(&＃39;start to writing data to OBS. output_path&＃61;%s&＃39; % output_path)if data_frame is None:logging.warning(&＃39;data_frame is None.&＃39;)returnif output_path is None or output_path.endswith(&＃39;s3://&＃39;):logging.error(&＃39;output_path is None or not endswith s3://&＃39;)returnif not os.path.exists(output_path):os.makedirs(output_path)data_frame.to_csv(&＃39;out.csv&＃39;, encoding&＃61;&＃39;utf-8&＃39;, index&＃61;False) # write to localmox.file.copy(&＃39;out.csv&＃39;, os.path.join(output_path, &＃39;out.csv&＃39;)) # copy to obslogging.info(&＃39;success write data to obs. output_file&＃61;%s&＃39; % os.path.join(output_path, &＃39;out.csv&＃39;))if __name__ &＃61;&＃61; &＃39;__main__&＃39;:logging.info(&＃39;start to running......&＃39;)# parse running argumentargs, un_known &＃61; argument_init()# Read from obsdataframe &＃61; read_from_obs(args.input_path)if dataframe is None:dataframe &＃61; pd.DataFrame({&＃39;col1&＃39;: [1, 2], &＃39;col2&＃39;: [3, 4]})# start call your own process function# dataframe &＃61; data_process(dataframe)# end call your own process function# Write to obswrite_to_obs(dataframe, args.output_path)logging.info(&＃39;finish.....&＃39;)


根据实际情况修改main.py代码&＃xff0c;在代码中是可以传递参数的哈&＃xff1b;然后点击“运行”

来到如下页面&＃xff1a;

补充一下&＃xff1a;

1&＃xff09;执行引擎有如下几种

2&＃xff09;如果main.py中不需要外部参数&＃xff0c;或者有默认的参数不需要改变&＃xff0c;可以不用传递参数。

完成上面的设置后&＃xff0c;点击“提交运行”&＃xff0c;看到状态是在运行中&＃xff0c;等待一会就看到运行成功啦。

好啦&＃xff0c;现在完成网络AI模型开发的数据入湖、数据处理。

友情链接&＃xff1a; 【华为云-上云之路】网络AI模型开发&＃xff08;1&＃xff09;创建IAM用户、订阅AI服务

福利&＃xff1a;华为云--公开课程《网络AI模型开发》

地址&＃xff1a;7天玩转网络AI模型开发

华为云官网出品&＃xff0c;有视频&＃xff0c;有课件&＃xff0c;都是免费的。