IDEA构建Spark编程环境和用Scala实现PageRank算法

简介: 写在前边的话       以前在学习编写mapreduce时,由于没有shell,就是首先在eclipse里配置环境,利用eclipse的强大功能进行编写,调试,编译,最终打包到集群上运行,同样面对Spark的时候,虽然spark提供了强大的shell 脚本能力,但对于定期或者处理时间很长...

写在前边的话

      以前在学习编写mapreduce时,由于没有shell,就是首先在eclipse里配置环境,利用eclipse的强大功能进行编写,调试,编译,最终打包到集群上运行,同样面对Spark的时候,虽然spark提供了强大的shell 脚本能力,但对于定期或者处理时间很长的程序来说并不适合使用shell,所以这里我选用了强大IDEA 作为spark的开发环境

      环境说明(点击下载):Java1.8.1_101    /    Scala 2.11.8    /     Intellij DEA 2.16.2   /  Spark 1.6.2

     10分钟 帮你打开Scala的编程大门:点击阅读 

    注意事项(我掉过的坑):本地Scala编程时,注意环境与集群的一致性,由于我的集群是spark1.6.2,scala2.10.x,java1.7.x,而本地环境java和scala都比集群高了一个版本,所以本地打包在集群上运行时就会出现版本不匹配的错误(如果用到java时,也是一样的),这一点大家要十分注意

一:IDEA构建Spark编程环境

       部署参考文章:使用IntelliJ IDEA编写SparkPi直接在Spark中运行

      需要注意的有两个地方

      1:在官方给的example中需要加入两行代码

    conf.setMaster("spark://192.168.48.130:7077")     //指定你的spark集群

    spark.addJar("/home/master/SparkApp/SparkTest.jar")   //指明位置
           由于我是在本地打包好的jar,所以上传jar到linux下时,必须保证位置与 代码中的一致        

       2:在提交jar包时,出现错误

 

           可以看出是17行的问题,原代码中17行:

val slices = if (args.length > 0) args(0).toInt else 2
        所以在运行jar包时需要指定该参数,官网给出的样例是这样的( http://spark.apache.org/docs/latest/

./bin/run-example SparkPi 10
        所以这里要指定数目

/opt/spark/bin/spark-submit /home/master/SparkApp/SparkApp.jar 10  --class "SparkApp"
        最终的运行结果如下



          此时我们在看Spark的Web界面监控


二:Spark执行PageRank算法

         PageRank算法解析参考:点击阅读

         PageRank的MapReduce实现参考:点击阅读

                                                     

        Shell 运行如下:

scala> val links = sc.parallelize(
     |  Array(
     |   ('A', Array('D')), 
     |   ('B', Array('A')), 
     |   ('C', Array('A', 'B')), 
     |   ('D', Array('A', 'C'))
     |  )
     | )
links: org.apache.spark.rdd.RDD[(Char, Array[Char])] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> var ranks = sc.parallelize(
     |  Array(
     |   ('A', 1.0), 
     |   ('B', 1.0), 
     |   ('C', 1.0), 
     |   ('D', 1.0)
     |  )
     | )
ranks: org.apache.spark.rdd.RDD[(Char, Double)] = ParallelCollectionRDD[1] at parallelize at <console>:24

scala> for(i <- 1 to 6){
     |     val joinRdd = links.join(ranks)
     |     val contribsRdd = joinRdd.flatMap{
     |       case(srcURL, (links, rank)) => links.map(destURL => (destURL, rank / links.size))
     |     }
     |     ranks = contribsRdd.reduceByKey(_ + _).mapValues(0.15 + _ * 0.85)
     |     ranks.take(4).foreach(println)
     |     println()
     | }

         六次迭代的结果:

        

      代码注释:

//图的初始化
val links = sc.parallelize(
 Array(
  ('A', Array('D')), 
  ('B', Array('A')), 
  ('C', Array('A', 'B')), 
  ('D', Array('A', 'C'))
 )
)

//PR值的初始化
//这里可以用 var ranks  = links.mapValues(_=> 1.0)代替
var ranks = sc.parallelize(
 Array(
  ('A', 1.0), 
  ('B', 1.0), 
  ('C', 1.0), 
  ('D', 1.0)
 )
)

//6 为循环次数,这里可以自己设置
for(i <- 1 to 6){
    val joinRdd = links.join(ranks)    //连接两个rdd
    //计算来自其他网页的PR 贡献值
    val contribsRdd = joinRdd.flatMap{
      // 注意这里的links为模式匹配得到的值, 类型为Array[Char], 并非前面的ParallelCollectionRDD
      case(srcURL, (links, rank)) => links.map(destURL => (destURL, rank / links.size))
    }
    //ranks进行更新
    ranks = contribsRdd.reduceByKey(_ + _).mapValues(0.15 + _ * 0.85)
    //打印出ranks的值
    ranks.take(4).foreach(println)
    println()   //换行,便于观察
}

           打包PageRank算法在Spark集群上运行(Jar包下载:github

/opt/spark/bin/spark-submit /home/master/SparkApp/Spark.jar --class "PageRank"

          运行结果如下:

         

         可以看到和Shell脚本运行的结果是一样的

相关文章
|
14天前
|
机器学习/深度学习 算法 搜索推荐
编程之舞:探索算法的优雅与力量
【6月更文挑战第10天】在软件的世界里,算法是构筑数字宇宙的基石。它们如同精心编排的舞蹈,每一个步骤都充满着逻辑的美感和解决问题的力量。本文将带领读者走进算法的世界,一起感受那些精妙绝伦的编程思想如何转化为解决现实问题的钥匙。
15 3
|
1天前
|
人工智能 算法 搜索推荐
Java算法编程详解和程序实例
Java算法编程详解和程序实例
|
4天前
|
分布式计算 资源调度 Java
Scala+Spark+Hadoop+IDEA实现WordCount单词计数,上传并执行任务(简单实例-下)
Scala+Spark+Hadoop+IDEA实现WordCount单词计数,上传并执行任务(简单实例-下)
13 0
|
4天前
|
分布式计算 Hadoop Scala
Scala +Spark+Hadoop+Zookeeper+IDEA实现WordCount单词计数(简单实例-上)
Scala +Spark+Hadoop+Zookeeper+IDEA实现WordCount单词计数(简单实例-上)
8 0
|
4天前
|
Java Scala Maven
Intellij IDEA+Maven+Scala第一个程序
Intellij IDEA+Maven+Scala第一个程序
7 0
|
15天前
|
算法 安全 数据挖掘
解锁编程之门:数论在算法与加密中的实用应用
解锁编程之门:数论在算法与加密中的实用应用
|
30天前
|
SQL 分布式计算 Java
IDEA 打包 Spark 项目 POM 文件依赖
这是一个 Maven POM 示例,用于构建一个使用 Spark 与 Hive 的项目,目标是将数据从 Hive 导入 ClickHouse。POM 文件设置了 Scala 和 Spark 的依赖,包括 `spark-core_2.12`, `spark-sql_2.12`, 和 `spark-hive_2.12`。`maven-assembly-plugin` 插件用于打包,生成包含依赖的和不含依赖的两种 JAR 包。`scope` 说明了依赖的使用范围,如 `compile`(默认),`provided`,`runtime`,`test` 和 `system`。
|
1月前
|
SQL 存储 分布式计算
在scala中使用spark
在scala中使用spark
25 0
|
2天前
|
算法
基于GA遗传优化的混合发电系统优化配置算法matlab仿真
**摘要:** 该研究利用遗传算法(GA)对混合发电系统进行优化配置,旨在最小化风能、太阳能及电池储能的成本并提升系统性能。MATLAB 2022a用于实现这一算法。仿真结果展示了一系列图表,包括总成本随代数变化、最佳适应度随代数变化,以及不同数据的分布情况,如负荷、风速、太阳辐射、弃电、缺电和电池状态等。此外,代码示例展示了如何运用GA求解,并绘制了发电单元的功率输出和年变化。该系统原理基于GA的自然选择和遗传原理,通过染色体编码、初始种群生成、适应度函数、选择、交叉和变异操作来寻找最优容量配置,以平衡成本、效率和可靠性。
|
3天前
|
机器学习/深度学习 算法
基于鲸鱼优化的knn分类特征选择算法matlab仿真
**基于WOA的KNN特征选择算法摘要** 该研究提出了一种融合鲸鱼优化算法(WOA)与K近邻(KNN)分类器的特征选择方法,旨在提升KNN的分类精度。在MATLAB2022a中实现,WOA负责优化特征子集,通过模拟鲸鱼捕食行为的螺旋式和包围策略搜索最佳特征。KNN则用于评估特征子集的性能。算法流程包括WOA参数初始化、特征二进制编码、适应度函数定义(以分类准确率为基准)、WOA迭代搜索及最优解输出。该方法有效地结合了启发式搜索与机器学习,优化特征选择,提高分类性能。

热门文章

最新文章