云原生企业级数据湖

网络规划

请参考表格中的说明和方案默认示例值为每个规划项做详细规划并在实际部署时将默认示例值修改为您的实际规划。

规划云资源

请参考表格中的说明和方案默认示例值为每个规划项做详细规划并在实际部署时将默认示例值修改为您的实际规划。

1. 创建专有网络VPC和交换机

您需要创建1个专有网络和1个交换机。

1. 登录专有网络管理控制台。

2. 在顶部菜单栏，选择华东2（上海）地域。

3. 在左侧导航栏，单击专有网络。

4. 在专有网络页面，单击创建专有网络。

5. 在创建专有网络页面上，配置1个专有网络和1台交换机。

   项目	说明	示例值
   VPC名称	建议您在部署过程中新建一个VPC作为本方案的专有网络。部署过程中填写VPC名称即可创建对应名称的VPC。 长度为2~128个字符，以英文字母或中文开头，可包含数字、下划线（_）和连字符（-）。	VPC_SH
   IPv4网段	在创建VPC时，您必须按照无类域间路由块（CIDR block）的格式为您的专有网络划分私网网段。阿里云VPC支持的网段信息请参见专有网络组成部分。 在网络规划时可以按照管理网段-开发网段-测试网段-生产网段等规则做好规划。网段一旦投入使用，调整过程复杂，因此规划十分重要。	192.168.0.0/16
   Vswitch名称	建议您在部署过程中在新建的VPC内创建虚拟交换机。部署过程中填写交换机名称即可创建对应名称的虚拟交换机。 长度为2~128个字符，以英文字母或中文开头，可包含数字、下划线（_）和连字符（-）。	vsw_001
   可用区	在规划的地域内选择1个可用区，1台虚拟交换机分别部署在1个可用区。建议选择排序靠后的，一般此类可用区较新。新可用区资源更充沛，新规格也会在新的可用区优先上线。	可用区 I
   IPv4网段	每台虚拟交换机需要一个IPv4网段	vsw_001：192.168.1.0/24

2. 创建安全组

接下来您需要创建1个安全组，用于限制该专有网络VPC下的交换机的网络流入和流出。

1. 登录ECS管理控制台。

2. 在顶部菜单栏，选择华东2（上海）地域。

3. 在左侧导航栏，选择网络与安全>安全组。

4. 在安全组页面，单击创建安全组。

5. 在创建安全组页面，创建1个安全组。

   项目	说明	示例值
   名称	设置安全组的名称。	SecurityGroup_EMR
   网络	选择之前规划的专有网络VPC。	VPC_SH
   安全组类型	本方案需从公网拉取博客网站服务镜像，因此选择普通安全组，以实现公网出方向所有地址可访问。实际部署时，您建议选择安全性更高的企业级安全组。	普通安全组
   端口	测试环境可为全部对象放行全部端口。实际生产环境，请只为独享资源组放行全部端口。	1/65535

3. 创建EMR集群

接下来您需要创建1个EMR集群，用于提供大数据计算引擎。

1. 登录EMR on ECS控制台。

2. 在顶部菜单栏处，选择华东2（上海）地域。

3. 在EMR on ECS页面，单击创建集群。

4. 在创建集群页面，参考下表创建1个集群，其他保持默认值。

   类型	项目	说明	示例值
   软件配置	地域	集群节点ECS实例所在的物理位置。重要集群创建后，无法更改地域，请谨慎选择。	华东2（上海）
   	业务场景	默认为新版数据湖。	新版数据湖
   	产品版本	选择当前最新的软件版本。	EMR-5.13.0
   	可选服务	根据您的实际需求选择组件，被选中的组件会默认启动相关的服务进程。	HADOOP-COMMON HDFS Hive Spark Tez YARN
   硬件配置	付费类型	在测试场景下，建议使用按量付费，测试正常后可以释放该集群，再新建一个包年包月的生产集群正式使用。	按量付费
   	可用区	集群创建后，无法直接更改可用区，请谨慎选择。	可用区 I
   	专有网络	选择对应区域下的专有网络。	VPC_SH/vpc-bp1e49hpm1mcluyij4***
   	交换机	选择在对应专有网络下可用区的交换机。	vsw_001/vsw-bp12ic8yl36r2k0ne7***
   	默认安全组	EMR目前只支持普通安全组，不支持企业安全组。安全组提供类似虚拟防火墙功能，用于设置集群节ECS实例的网络访问控制，是重要的安全隔离手段。	sg_seurity/sg-bp17bpb8ucnsilrh5***
   基础配置	集群名称	集群的名字，长度限制为1~64个字符，仅可使用中文、字母、数字、短划线（-）和下划线（_）。	Emr-DataLake
   	身份凭证	用于远程登录集群的Master节点。	Hello123****
   	登录密码和确认密码	请记录该配置，登录集群时您需要输入该密码。	Hello123****

4. 创建对象存储OSS Bucket

接下来您需要创建1个对象存储OSS Bucket，用于存储待分析的数据。

1. 登录OSS管理控制台。

2. 在顶部菜单栏，选择华东2（上海）地域。

3. 在左侧导航栏，选择Bucket列表。

4. 在Bucket列表页面，单击创建Bucket。

5. 在创建Bucket面板，创建1个Bucket。

   项目	说明	示例值
   Bucket名称	Bucket 命名规范： 命名长度为3~63个字符。 只允许小写字母、数字、短横线（-），且不能以短横线开头或结尾。 Bucket名称在OSS范围内必须全局唯一。	example-bucket-data-lake
   地域属性	本方案以华东2（上海）为例。	华东2（上海）
   存储类型	数据会经常被访问，因此需要确保高可靠、高可用、高性能。	标准存储
   读写权限	对数据的所有访问操作需要进行身份验证。	私有

1. 上传待分析的数据

将待分析的数据上传到OSS。

1. 下载待分析的文件。

   示例文件：The_Sorrows_of_Young_Werther.txt

2. 上传待分析的文件到OSS。

   1. 登录OSS管理控制台。

   2. 在顶部菜单栏，选择华东2（上海）地域。

   3. 在左侧导航栏，选择Bucket列表。

   4. Bucket列表页面，单击目标Bucket。

   5. 在左侧导航栏，选择文件管理 > 文件列表。

   6. 在文件列表页面，单击新建目录，新建inputs目录，然后单击上传文件，将待分析的文件上传到inputs目录下。

2. 编写数据分析作业

基于Hadoop的MapReduce框架在本地编写MR作业。

1. 在本地安装Maven和Java环境。

   本示例中Maven是3.0版本，Java是1.8版本。

2. 执行以下命令生成工程框架。

   例如，您的工程开发根目录是D:/workspace。

   mvn archetype:generate -DgroupId=com.aliyun.emr.hadoop.examples -DartifactId=wordcountv2 -DarchetypeArtifactId=maven-archetype-quickstart -DinteractiveMode=false

   通过以上命令会自动生成一个空的Sample工程位于D:/workspace/wordcountv2（和您指定的artifactId一致），里面包含一个简单的pom.xml和App类（类的包路径和您指定的groupId一致）。

3. 加入Hadoop依赖。

   使用IDE打开Sample工程，编辑pom.xml文件，当Hadoop是2.8.5版本时，需要添加如下内容。

   <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-mapreduce-client-common</artifactId>
    <version>2.8.5</version>
   </dependency>
   <dependency>
    <groupId>org.apache.hadoop</groupId>
    <artifactId>hadoop-common</artifactId>
    <version>2.8.5</version>
   </dependency>

4. 编写代码。

   1. 在com.aliyun.emr.hadoop.examples中和App类平行的位置添加新类EMapReduceOSSUtil。

      package com.aliyun.emr.hadoop.examples;
       import org.apache.hadoop.conf.Configuration;
       public class EMapReduceOSSUtil {
       private static String SCHEMA = "oss://";
       private static String EPSEP = ".";
       private static String HTTP_HEADER = "http://";
       /**
       * complete OSS uri
       * convert uri like: oss://bucket/path to oss://bucket.endpoint/path
       * ossref do not need this
       *
       * @param oriUri original OSS uri
       */
       public static String buildOSSCompleteUri(String oriUri, String endpoint) {
       if (endpoint == null) {
       System.err.println("miss endpoint");
       return oriUri;
       }
       int index = oriUri.indexOf(SCHEMA);
       if (index == -1 || index != 0) {
       return oriUri;
       }
       int bucketIndex = index + SCHEMA.length();
       int pathIndex = oriUri.indexOf("/", bucketIndex);
       String bucket = null;
       if (pathIndex == -1) {
       bucket = oriUri.substring(bucketIndex);
       } else {
       bucket = oriUri.substring(bucketIndex, pathIndex);
       }
       StringBuilder retUri = new StringBuilder();
       retUri.append(SCHEMA)
       .append(bucket)
       .append(EPSEP)
       .append(stripHttp(endpoint));
       if (pathIndex > 0) {
       retUri.append(oriUri.substring(pathIndex));
       }
       return retUri.toString();
       }
       public static String buildOSSCompleteUri(String oriUri, Configuration conf) {
       return buildOSSCompleteUri(oriUri, conf.get("fs.oss.endpoint"));
       }
       private static String stripHttp(String endpoint) {
       if (endpoint.startsWith(HTTP_HEADER)) {
       return endpoint.substring(HTTP_HEADER.length());
       }
       return endpoint;
       }
       }

   2. 在com.aliyun.emr.hadoop.examples中和App类平行的位置添加新类WordCount2.java，并修改其中的访问密钥和Endpoint。其中，Endpoint使用OSS Bucket的内网域名。

      package com.aliyun.emr.hadoop.examples;
      import java.io.BufferedReader;
      import java.io.FileReader;
      import java.io.IOException;
      import java.net.URI;
      import java.util.ArrayList;
      import java.util.HashSet;
      import java.util.List;
      import java.util.Set;
      import java.util.StringTokenizer;
      import org.apache.hadoop.conf.Configuration;
      import org.apache.hadoop.fs.Path;
      import org.apache.hadoop.io.IntWritable;
      import org.apache.hadoop.io.Text;
      import org.apache.hadoop.mapreduce.Job;
      import org.apache.hadoop.mapreduce.Mapper;
      import org.apache.hadoop.mapreduce.Reducer;
      import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
      import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
      import org.apache.hadoop.mapreduce.Counter;
      import org.apache.hadoop.util.GenericOptionsParser;
      import org.apache.hadoop.util.StringUtils;
      
      
      public class WordCount2 {
          public static class TokenizerMapper
                  extends Mapper<Object, Text, Text, IntWritable>{
              static enum CountersEnum { INPUT_WORDS }
              private final static IntWritable one = new IntWritable(1);
              private Text word = new Text();
              private boolean caseSensitive;
              private Set<String> patternsToSkip = new HashSet<String>();
              private Configuration conf;
              private BufferedReader fis;
              @Override
              public void setup(Context context) throws IOException,
                      InterruptedException {
                  conf = context.getConfiguration();
                  caseSensitive = conf.getBoolean("wordcount.case.sensitive", true);
                  if (conf.getBoolean("wordcount.skip.patterns", true)) {
                      URI[] patternsURIs = Job.getInstance(conf).getCacheFiles();
                      System.out.print(conf.toString());
                      System.out.print(patternsURIs);
                      if (patternsURIs == null) {
                          return;
                      }
                      for (URI patternsURI : patternsURIs) {
                          Path patternsPath = new Path(patternsURI.getPath());
                          String patternsFileName = patternsPath.getName().toString();
                          parseSkipFile(patternsFileName);
                      }
                  }
              }
              private void parseSkipFile(String fileName) {
                  try {
                      fis = new BufferedReader(new FileReader(fileName));
                      String pattern = null;
                      while ((pattern = fis.readLine()) != null) {
                          patternsToSkip.add(pattern);
                      }
                  } catch (IOException ioe) {
                      System.err.println("Caught exception while parsing the cached file '"
                              + StringUtils.stringifyException(ioe));
                  }
              }
              @Override
              public void map(Object key, Text value, Context context
              ) throws IOException, InterruptedException {
                  String line = (caseSensitive) ?
                          value.toString() : value.toString().toLowerCase();
                  for (String pattern : patternsToSkip) {
                      line = line.replaceAll(pattern, "");
                  }
                  StringTokenizer itr = new StringTokenizer(line);
                  while (itr.hasMoreTokens()) {
                      word.set(itr.nextToken());
                      context.write(word, one);
                      Counter counter = context.getCounter(CountersEnum.class.getName(),
                              CountersEnum.INPUT_WORDS.toString());
                      counter.increment(1);
                  }
              }
          }
          public static class IntSumReducer
                  extends Reducer<Text,IntWritable,Text,IntWritable> {
              private IntWritable result = new IntWritable();
              public void reduce(Text key, Iterable<IntWritable> values,
                                 Context context
              ) throws IOException, InterruptedException {
                  int sum = 0;
                  for (IntWritable val : values) {
                      sum += val.get();
                  }
                  result.set(sum);
                  context.write(key, result);
              }
          }
          public static void main(String[] args) throws Exception {
              Configuration conf = new Configuration();
              conf.set("fs.oss.accessKeyId", "<yourAccessKeyID>");
              conf.set("fs.oss.accessKeySecret", "<yourAccessKeySecret>");
              conf.set("fs.oss.endpoint","<yourOSSEndpoint>");
              GenericOptionsParser optionParser = new GenericOptionsParser(conf, args);
              String[] remainingArgs = optionParser.getRemainingArgs();
              if (!(remainingArgs.length != 2 || remainingArgs.length != 4)) {
                  System.err.println("Usage: wordcount <in> <out> [-skip skipPatternFile]");
                  System.exit(2);
              }
              Job job = Job.getInstance(conf, "word count");
              job.setJarByClass(WordCount2.class);
              job.setMapperClass(TokenizerMapper.class);
              job.setCombinerClass(IntSumReducer.class);
              job.setReducerClass(IntSumReducer.class);
              job.setOutputKeyClass(Text.class);
              job.setOutputValueClass(IntWritable.class);
              List<String> otherArgs = new ArrayList<String>();
              for (int i=0; i < remainingArgs.length; ++i) {
                  if ("-skip".equals(remainingArgs[i])) {
                      job.addCacheFile(new Path(EMapReduceOSSUtil.buildOSSCompleteUri(remainingArgs[++i], conf)).toUri());
                      job.getConfiguration().setBoolean("wordcount.skip.patterns", true);
                  } else {
                      otherArgs.add(remainingArgs[i]);
                  }
              }
              FileInputFormat.addInputPath(job, new Path(EMapReduceOSSUtil.buildOSSCompleteUri(otherArgs.get(0), conf)));
              FileOutputFormat.setOutputPath(job, new Path(EMapReduceOSSUtil.buildOSSCompleteUri(otherArgs.get(1), conf)));
              System.exit(job.waitForCompletion(true) ? 0 : 1);
          }
      }

5. 在工程的目录下，执行如下命令，编译并打包文件。

   mvn clean package -DskipTests

   您可以在工程目录的target目录下看到名称为wordcountv2-1.0-SNAPSHOT.jar的JAR包。

3. 创建并运行数据分析作业

在EMR集群中创建MR作业并运行。

1. 登录EMR on ECS控制台。

2. 在顶部菜单栏处，选择华东2（上海）地域。

3. 在EMR on ECS页面，单击创建的集群的ID。

4. 选择节点管理页签，单击emr-master节点组左侧的加号，然后单击master-1-1节点的ID。

5. 在云服务器ECS的实例详情页面的基本信息中，单击远程连接，然后根据页面提示登录ECS。

6. 在顶部菜单栏，选择文件 > 打开新文件管理 > root，单击上传文件，根据页面提示上传wordcountv2-1.0-SNAPSHOT.jar。

7. 在root目录下执行以下命令创建并运行数据分析作业。

   spark-submit --class com.aliyun.emr.hadoop.examples.WordCount2 wordcountv2-1.0-SNAPSHOT.jar oss://example-bucket-data-lake/inputs oss://example-bucket-data-lake/outputs

4. 上传训练数据

将待训练的数据上传到OSS。

1. 下载并解压待训练的文件。

   示例文件：hymenoptera_data.zip

2. 上传待训练的文件夹到OSS。

   1. 登录OSS管理控制台。

   2. 在顶部菜单栏，选择华东2（上海）地域。

   3. 在左侧导航栏，选择Bucket列表。

   4. Bucket列表页面，单击目标Bucket。

   5. 在左侧导航栏，选择文件管理 > 文件列表。

   6. 在文件列表页面，然后单击上传文件，选择扫描文件夹，然后根据页面提示，上传待训练的文件夹。文件夹中的文件数量较多，请勿中途关闭上传页面，以免上传失败。

5. 创建数据集

使用OSS中的数据创建1个数据集。

1. 登录PAI控制台。

2. 在左侧导航栏，选择AI资产管理 > 数据集。

3. 在数据集管理页面，单击创建数据集。

4. 在创建数据集面板，参考下表填写参数，然后单击提交。

   项目	说明	示例值
   数据集名称	本文以常用数据集hymenoptera_data为例。该数据集用于图像分类任务，特别是昆虫分类。它包含了两个类别的昆虫图像：蚂蚁和蜜蜂。	hymenoptera_data
   从阿里云云存储创建	选择上一步将训练数据上传到OSS Bucket后生成的文件路径。	oss://oss-pai-demo.oss-cn-shanghai.aliyuncs.com/hymenoptera_data/

6. 创建并运行模型训练任务

使用数据集训练1个深度学习模型。

1. 在PAI控制台的左侧导航栏，选择模型开发与训练 > 分布式训练（DLC）。

2. 在分布式训练页面，单击新建任务，参考下表填写参数，然后单击提交。

   项目	说明	示例值
   任务名称	本文要训练的模型是基于ResNet18架构进行微调的，用于图像分类任务。	resnet18-ft-hymenoptera-classifier
   节点镜像	本文以pytorch社区镜像为例。	pytorch-training:1.1.12-cpu-py39-cu110-ubuntu20.04
   框架	本文以深度学习框架PyTorch为例。PyTorch提供了用于构建和训练神经网络的高级API和工具，是一个用于进行机器学习和深度学习的强大工具。	PyTorch
   数据集配置	选择创建的数据集进行挂载。	hymenoptera_data /mnt/data
   执行命令	示例执行命令使用wget下载名为main.py的Python脚本文件，并通过python命令将输入目录/mnt/data和输出目录/mnt/data/作为命令行参数传递给该脚本。	wget https://pai-public-data.oss-cn-beijing.aliyuncs.com/hol-pytorch-transfer-cv/main.py && python main.py -i /mnt/data -o /mnt/data/