开发程序

提供1个开发应用程序案例，基于图分析算法加速库中的TC算法。

将工程中“src/main和src/test”目录下的“java”文件夹重命名为“scala”，“java”目录上右击，依次选择“Refactor”、“Rename”，然后输入“scala”。

在根目录下pom文件中添加依赖，将以下内容复制替换到pom.xml中。

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchemainstance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>com.bigdata</groupId>
    <artifactId>kal_examples_2.11</artifactId>
    <version>0.1</version>
    <name>${project.artifactId}</name>
    <inceptionYear>2020</inceptionYear>
    <packaging>jar</packaging>
    
    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <encoding>UTF-8</encoding>
        <scala.version>2.11.8</scala.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-graphx_2.11</artifactId>
            <version>2.3.2</version>
        </dependency>
        <dependency>
            <groupId>it.unimi.dsi</groupId>
            <artifactId>fastutil</artifactId>
            <version>8.3.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark.graphx.lib</groupId>
            <artifactId>sophon-graph-kernel-client_2.11</artifactId>
            <version>1.2.0</version>
        </dependency>
    </dependencies>
    <build>
        <sourceDirectory>src/main/scala</sourceDirectory>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.0</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                        </goals>
                        <configuration>
                            <args>
                                <arg>-dependencyfile</arg>
                                <arg>${project.build.directory}/.scala_dependencies</arg>
                            </args>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

在根目录下新建lib文件夹，右键单击“kal_examples > > New > Directory”。
将软件编译获取得到的sophon-graph-kernel-client_2.11-1.2.0.jar包放入新建的lib文件中。
将sophon-graph-kernel-client_2.11-1.2.0.jar安装到本地仓，单击右侧边框“Maven > M”，输入“install:install-file -DgroupId=org.apache.spark.graphx.lib DartifactId=sophon-graph-kernel-client_2.11 -Dversion=1.2.0 -Dfile=lib/sophongraph-kernel-client_2.11-1.2.0.jar -Dpackaging=jar”，按“Enter”。
在新建工程中，在“src/main/scala/”目录下，新建package“com.bigdata.examples”，右键单击“scala > New > Package”。

输入“com.bigdata.examples”，单击“OK”。

在6中创建的“com.bigdata.examples”package下新建TCRunner.scala文件。右击“com.bigdata.examples”选择“New > File”。

输入“TCRunner.scala”，单击“OK”。

将以下代码复制到TCRunner.scala文件中：

package com.bigdata.examples
import org.apache.spark.graphx
import org.apache.spark.graphx.Graph
import org.apache.spark.graphx.lib.TriangleCount
import org.apache.spark.rdd.RDD
import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
object TCRunner {
def main(args: Array[String]): Unit = {
val numPartitions = 200 // 分区数为200
val split = " " // 分隔符为空格

// 创建Spark运行环境
val sparkConf = new SparkConf().setAppName(s"TC").setMaster("yarn") val sc = new SparkContext(sparkConf)
// 记录程序运行开始时间
val startTime = System.currentTimeMillis()
val inputData = sc.textFile("hdfs:///tmp/graph_data/graph500-23.e", numPartitions) // 给定输入路径和分区数
.flatMap(line => {
if (line.startsWith("#")) { // 数据行以#开始，跳过读取这一行
Iterator.empty
} else {
val x = line.split(split) // 按照分隔符读取每一行数据
if (x.length < 2) { // 如果有数据缺失，则跳过读取这一行
Iterator.empty
} else {
val node1 = x(0) // 第一个节点数据
val node2 = x(1) // 第二个节点数据
Iterator((node1, node2)) // Scala迭代器，将每一组节点连接起来
}
}
})
// 进行数据类型转换，并按照大小顺序排列节点位置
val inputRdd = inputData.flatMap(x => {
if (x._1.toLong == x._2.toLong) { // 排除起始节点都是自身的自环节点
Iterator.empty
} else if (x._1.toLong > x._2.toLong) {
Iterator((x._2.toLong, x._1.toLong)) // 始终保持第一个节点数值小于第二个节点的数值
} else {
Iterator((x._1.toLong, x._2.toLong))
}
})
// 用Spark框架构建图
val graph = Graph.fromEdgeTuples(inputRdd, 0)
var result: RDD[(graphx.VertexId, Int)] = null
// 调用TC算法run接口，计算结果
result = TriangleCount.run(graph).vertices.repartition(numPartitions.toInt)
// 保存数据到指定输出路径
result.map(f => f._1 + "," + f._2).saveAsTextFile("hdfs:///tmp/graph_result/graph500-23-result") // 将结果输出到指定路径
// 记录程序花费时间
val costTime = (System.currentTimeMillis() - startTime) / 1000.0
println("Triangle Counting is finished , and costTime = " + costTime + " 's")
}
}

图1 文件目录结构

单击右侧边框“Maven > M”，输入“package”，按“Enter”，进行项目打包，在“target\”目录中生成kal_examples_2.11-0.1.jar。

父主题： 样例工程