1010      阿里云  技术社区[云栖]

图模型功能介绍__图模型_大数据计算服务-阿里云

MaxCompute 客户端提供一个Jar命令用于运行 MaxCompute GRAPH作业，其使用方式与 MapReduce中的Jar命令 相同，这里仅作简要介绍：

Usage: jar [<GENERIC_OPTIONS>] <MAIN_CLASS> [ARGS]
    -conf <configuration_file>         Specify an application configuration file
    -classpath <local_file_list>       classpaths used to run mainClass
    -D <name>=<value>                  Property value pair, which will be used to run mainClass
    -local                             Run job in local mode
    -resources <resource_name_list>    file/table resources used in graph, seperate by comma

其中 < GENERIC_OPTIONS>包括(均为可选参数)：

• -conf <configuration file > ：指定JobConf配置文件；
• -classpath <local_file_list > : 本地执行时的classpath，主要用于指定main函数所在的jar包。大多数情况下，用户更习惯于将main函数与Graph作业编写在一个包中，例如：单源最短距离算法 ，因此，在执行示例程序时，-resources及-classpath的参数中都出现了用户的jar包，但二者意义不同，-resources引用的是Graph作业，运行于分布式环境中，而-classpath引用的是main函数，运行于本地，指定的jar包路径也是本地文件路径。包名之间使用系统默认的文件分割符作分割(通常情况下，windows系统是分号”;”，linux系统是冒号”:”)；
• -D <prop_name > = < prop_value > : 本地执行时，<mainClass > 的java属性，可以定义多个；
• -local：以本地模式执行Graph作业，主要用于程序调试；
• -resources <resource_name_list > : Graph作业运行时使用的资源声明。一般情况下，resource_name_list中需要指定Graph作业所在的资源名称。如果用户在Graph作业中读取了其他ODPS资源，那么，这些资源名称也需要被添加到resource_name_list中。资源之间使用逗号分隔，使用跨项目空间使用资源时，需要前面加上：PROJECT_NAME/resources/，示例：-resources otherproject/resources/resfile；

同时，用户也可以直接运行GRAPH作业的main函数直接将作业提交到 MaxCompute ，而不是通过 MaxCompute 客户端提交作业。以PageRank算法 为例:

public static void main(String[] args) throws IOException {
  if (args.length < 2)
    printUsage();
  GraphJob job = new GraphJob();
  job.setGraphLoaderClass(PageRankVertexReader.class);
  job.setVertexClass(PageRankVertex.class);
  job.addInput(TableInfo.builder().tableName(args[0]).build());
  job.addOutput(TableInfo.builder().tableName(args[1]).build());
  // 将作业中使用的资源添加到cache resource，对应于jar命令中 -resources 和 -libjars 中指定的资源
  job.addCacheResource("mapreduce-examples.jar");
  // 将使用的jar及其他文件添加到class cache resource，对应于jar命令中 -libjars 中指定的资源
  job.addCacheResourceToClassPath("mapreduce-examples.jar");
  // 设置console中，odps_config.ini对应的配置项，使用时替换为自己的配置
  OdpsConf.getInstance().setProjName("project_name");
  OdpsConf.getInstance().setEndpoint("end_point");
  OdpsConf.getInstance().setAccessId("access_id");
  OdpsConf.getInstance().setAccessKey("access_key");
  // default max iteration is 30
  job.setMaxIteration(30);
  if (args.length >= 3)
    job.setMaxIteration(Integer.parseInt(args[2]));
  long startTime = System.currentTimeMillis();
  job.run();
  System.out.println("Job Finished in "
    + (System.currentTimeMillis() - startTime) / 1000.0 + " seconds");
}

GraphJob job = new GraphJob();
job.addInput(TableInfo.builder().tableName(“tblname”).build());  //表作为输入
job.addInput(TableInfo.builder().tableName(“tblname”).partSpec("pt1=a/pt2=b").build()); //分区作为输入
//只读取输入表的 col2 和 col0 列，在 GraphLoader 的 load 方法中，record.get(0) 得到的是col2列，顺序一致
job.addInput(TableInfo.builder().tableName(“tblname”).partSpec("pt1=a/pt2=b").build(), new String[]{"col2", "col0"});

GraphJob job = new GraphJob();
//输出表为分区表时需要给到最末一级分区
job.addOutput(TableInfo.builder().tableName("table_name").partSpec("pt1=a/pt2=b").build());
// 下面的参数 true 表示覆盖tableinfo指定的分区，即INSERT OVERWRITE语义，false表示INSERT INTO语义
job.addOutput(TableInfo.builder().tableName("table_name").partSpec("pt1=a/pt2=b").lable("output1").build(), true);

void addCacheResources(String resourceNames)
void addCacheResourcesToClassPath(String resourceNames)

public byte[] readCacheFile(String resourceName) throws IOException;
public Iterable<byte[]> readCacheArchive(String resourceName) throws IOException;
public Iterable<byte[]> readCacheArchive(String resourceName, String relativePath)throws IOException;
public Iterable<WritableRecord> readResourceTable(String resourceName);
public BufferedInputStream readCacheFileAsStream(String resourceName) throws IOException;
public Iterable<BufferedInputStream> readCacheArchiveAsStream(String resourceName) throws IOException;
public Iterable<BufferedInputStream> readCacheArchiveAsStream(String resourceName, String relativePath) throws IOException;