Spark算子-Scala版本头歌答案

4师傅 2023-08-26 原文

Spark算子--Scala版本

第1关 Spark算子--Scala版本

编程要求

根据提示，在右侧编辑器begin-end处补充代码，输出每个元素及其长度并去重。

测试说明

平台会对你编写的代码进行测试：

预期输出：

(an,2)` `(dog,3)` `(cat,3)

开始你的任务吧，祝你成功！

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object EduCoder1 {
    def main(args: Array[String]): Unit = {
    val conf =new SparkConf().setAppName("educoder1").setMaster("local")
    val sc=new SparkContext(conf)
    val rdd = sc.parallelize(List("dog","an","cat","an","cat"))
   
    
    /********** Begin **********/
    //第一步：通过获取rdd中每个元素的长度创建新的rdd1
    val rdd1=rdd.map(x=>x.length)

    
    //第二步：通过zip把rdd1和rdd组合创建rdd2
    
     val rdd2=rdd.zip(rdd1)

    //第三步：去重

    val rdd3=rdd2.distinct()
    
    //第四步：输出结果
    rdd3.foreach(println)
    
    
    /********** End **********/
  
    sc.stop()
  }

}

第2关：转换算子之flatMap和filter算子

编程要求

根据提示，在右侧编辑器begin-end处补充代码，输出个数大于一的单词。

测试说明

平台会对你编写的代码进行测试：

所给文件内容如下：

hello,world,hello,sparkgood,nice,good,do

预期输出： (hello,2) (good,2)

开始你的任务吧，祝你成功！

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object EduCoder2 {
    def main(args: Array[String]): Unit = {
    val conf =new SparkConf().setAppName("educoder2").setMaster("local")
    val sc=new SparkContext(conf)
    val rdd=sc.textFile("file:///root/step3_fils")
    

    /********** Begin **********/
     //对所给数据创建的rdd切割分词

    val rdd1=rdd.flatMap(t=>t.split(","))

    //每个单词计数为1

    val rdd2= rdd1.map(t=>(t,1))

    //对相同单词个数进行累加

    val rdd3=rdd2.reduceByKey(_+_)

    //过滤出单词个数大于一个的

    val rdd4= rdd3.filter(t=>t._2>1)

    //输出结果

    rdd4.foreach(println)
    
    /********** End **********/

    sc.stop()
  }

}

第3关：转换算子之reduceBykey和mapValues算子

编程要求

根据提示，在右侧编辑器begin-end处补充代码，某商店上午卖出10本 spark 书籍，每本50元，4本 Hadoop 书籍，每本40元，下午卖出20本 spark 书籍，每本40元，10本 Hadoop 书籍，每本30元。

现要求求出这两本书这一天销售的平均价格。

数据如下：

spark,10,50spark,40,25hadoop,5,40hadoop,10,25

测试说明

平台会对你编写的代码进行测试：

预期输出：

(spark,30)` `(hadoop,30)

开始你的任务吧，祝你成功！

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object EduCoder3 {
    def main(args: Array[String]): Unit = {
    
    val conf =new SparkConf().setAppName("educoder3").setMaster("local")
    val sc=new SparkContext(conf)
    
    /********** Begin **********/
     //通过给定数据通过序列化方式创建rdd

    val rdd = sc.parallelize(List(("spark",(10,50)),("hadoop",(5,40)),("hadoop",(10,25)),("spark",(40,25))))

    //求出一天收入总和以及出售本数

    val rdd2 = rdd.reduceByKey((x,y) => ((x._1*x._2)+(y._1*y._2), x._1+y._1))

    //求出每本平均售价

    val rdd3 = rdd2.mapValues(x => x._1 / x._2)

    //输出结果

    rdd3.foreach(println)
   
    /********** End **********/
   
    sc.stop

}
}

第4关：转化算子之groupByKey和sortByKey

编程要求

根据提示，在右侧编辑器begin-end处补充代码，对每人所学书籍本数分组并排序输出。

测试说明

平台会对你编写的代码进行测试：

所给数据说明：("Bob","spark") Bob：人名 spark：所学书籍

预期输出：

(Bob,3)(Candy,1)(Lily,1)

开始你的任务吧，祝你成功！

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object EduCoder4 {
    def main(args: Array[String]): Unit = {
    val conf =new SparkConf().setAppName("educoder4").setMaster("local")
    val sc=new SparkContext(conf)
   val rdd = sc.parallelize(List(("Bob","spark"),("Lily","hadoop"),("Candy","hive"),("Bob","hbase"),("Bob","hive")))
     
    /********** Begin **********/
     //根据姓名对所学书籍分组

    val rdd1= rdd.groupByKey()

    //求出每个人的书籍本数

    val rdd2= rdd1.mapValues(t=>t.toList.size)

    //根据姓名排序

    val rdd3= rdd2.sortByKey()

    //打印结果

    rdd3.foreach(println)
    
    /********** End **********/
    sc.stop()
  }

}

第5关：常见行动算子

编程要求

根据提示，在右侧编辑器begin-end处补充代码输出正确答案。

测试说明

预期输出：

4
dog
sun
an
dogsunancat
dog
sun
an
cat

开始你的任务吧，祝你成功！

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object EduCoder5 {
    def main(args: Array[String]): Unit = {
    val conf =new SparkConf().setAppName("educoder5").setMaster("local")
    val sc=new SparkContext(conf)
    val rdd = sc.parallelize(List("dog","sun","an","cat"))
     
    /********** Begin **********/
   
     //返回所给rdd的元素个数并输出

    val r1=rdd.count()

        println(r1)

    //返回rdd的前三个元素并输出

        val rdd1=rdd.take(3)

        rdd1.foreach(println)

    //累加rdd的所有元素并输出结果

        val r2=rdd.reduce(_+_)

        println(r2)

        //收集所有元素并且输出

       rdd.collect().foreach(println)
    
    /********** End **********/
   
    sc.stop()
  }

}

第6关：算子的综合使用案例

编程要求

有一份数据格式如下的文档：

日期，姓名，app,下载渠道，地区，版本号

2017-08-14,Lily,Facebook,360 Shop,NewYork,v1.0 2017-08-14,Bob,Facebook,Amazon Appstore,NewYork,v1.2 2017-08-14,Lily,Facebook,360 Shop,Washington,v1.2 2017-08-14,Lily,Facebook,Google Play Store,Washington,v2.0 2017-08-14,Candy,YouTube,app store,Chicago,v1.8 2017-08-14,Lily,Facebook,Google Play Store,Washington,v2.0 2017-08-14,Candy,YouTube,app store,Chicago,v1.9 2017-08-15,Candy,YouTube,app store,Chicago,v2.0 2017-08-15,Candy,YouTube,app store,Chicago,v2.3 2017-08-15,Lily,Facebook,360 Shop,NewYork,v2.0 2017-08-15,Bob,Facebook,Amazon Appstore,NewYork,v1.2 2017-08-15,Bob,Facebook,Amazon Appstore,NewYork,v1.5 2017-08-15,Candy,YouTube,app store,Chicago,v2.9

需求：不考虑地区，列出版本升级情况。

结果格式：日期，姓名，app，下载渠道，升级前版本，升级后版本。

例：数据：

2017-08-14,Lily,Facebook,360 Shop,NewYork,v1.0 2017-08-14,Lily,Facebook,360 Shop,Washington,v1.2 2017-08-14,Lily,Facebook,360 Shop,NewYork,v2.0

结果：

(2017-08-14,Lily,Facebook,360 Shop,v1.0,v1.2） (2017-08-14,Lily,Facebook,360 Shop,v1.2,v2.0）

测试说明

本实训目前是基于Spark单机模式的运行方式，完成整个评测流程所需时间较长，请耐心等待!

开始你的任务吧，祝你成功！

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object EduCoder {
  def main(args: Array[String]): Unit = {
    val conf =new SparkConf().setAppName("educoder").setMaster("local")
    val sc=new SparkContext(conf)
    val line=sc.textFile("file:///root/step1_fils")
    
    /********** Begin **********/
    //根据需求，去除城市字段

      val rdd1 = line.map(t => {

      val arr = t.split(",")

      ((arr(0), arr(1), arr(2), arr(3)), arr(5))

      })

      //按key分组，key是除城市字段和版本号字段``以外的所有字段，value是版本号

      val rdd2=rdd1.groupByKey()

      //过滤版本号重复的``(例：（v2.0,v2.0）)以及版本号只有一个的(例（v1.0）) 

      val rdd3=rdd2.mapValues(t=>t.toList.distinct).filter(t=>t._2.length>1)

      //拆分重新组合 例：(key,（v2.0,v2.5,v3.0))拆分成(key,（v2.0,v2.5）,（(key,(v2.5,v3.0)）)

      val rdd4= rdd3.mapValues(t => { 

      val tai = t.tail 

      t.zip(tai)

      })

      //按需求整理输出格式(例：（2017-08-14,Lily,Facebook,360  Shop,v1.2,v2.0）)

      val rdd5= rdd4.flatMap(t => {

      t._2.map(tp => {

      (t._1._1, t._1._2, t._1._3, t._1._4, tp._1, tp._2)

      })

      })

      //执行foreach操作，打印出结果

      rdd5.foreach(println)

    
    /********** End **********/
   
    sc.stop()
  }

}

有关Spark算子-Scala版本头歌答案的更多相关文章

ruby-on-rails - 项目升级后 Pow 不会更改 ruby 版本 - 2
我在我的Rails项目中使用Pow和powifygem。现在我尝试升级我的ruby版本(从1.9.3到2.0.0，我使用RVM)当我切换ruby版本、安装所有gem依赖项时，我通过运行railss并访问localhost:3000确保该应用程序正常运行以前，我通过使用pow访问http://my_app.dev来浏览我的应用程序。升级后，由于错误Bundler::RubyVersionMismatch:YourRubyversionis1.9.3,butyourGemfilespecified2.0.0，此url不起作用我尝试过的:重新创建pow应用程序重启pow服务器更新战俘
ruby-on-rails - 在 ruby .gemspec 文件中，如何指定依赖项的多个版本？ - 2
我正在尝试修改当前依赖于定义为activeresource的gem:s.add_dependency"activeresource","~>3.0"为了让gem与Rails4一起工作，我需要扩展依赖关系以与activeresource的版本3或4一起工作。我不想简单地添加以下内容，因为它可能会在以后引起问题:s.add_dependency"activeresource",">=3.0"有没有办法指定可接受版本的列表？~>3.0还是~>4.0？最佳答案根据thedocumentation,如果你想要3到4之间的所有版本，你可以这
ruby-on-rails - 如果我将 ruby 版本 2.5.1 与 rails 版本 2.3.18 一起使用会怎样？ - 2
如果我使用ruby版本2.5.1和Rails版本2.3.18会怎样？我有基于rails2.3.18和ruby1.9.2p320构建的rails应用程序，我只想升级ruby的版本，而不是rails，这可能吗？我必须面对哪些挑战？最佳答案 GitHub维护apublicfork它有针对旧Rails版本的分支，有各种变化，它们一直在运行。有一段时间，他们在较新的Ruby版本上运行较旧的Rails版本，而不是最初支持的版本，因此您可能会发现一些关于需要向后移植的有用提示。不过，他们现在已经有几年没有使用2.3了，所以充其量只能让更
ruby-on-rails - 获取 inf-ruby 以使用 ruby 版本管理器 (rvm) - 2
我安装了ruby版本管理器，并将RVM安装的ruby实现设置为默认值，这样'哪个ruby'显示'~/.rvm/ruby-1.8.6-p383/bin/ruby'但是当我在emacs中打开inf-ruby缓冲区时，它使用安装在/usr/bin中的ruby。有没有办法让emacs像shell一样尊重ruby的路径？谢谢! 最佳答案我创建了一个emacs扩展来将rvm集成到emacs中。如果您有兴趣，可以在这里获取:http://github.com/senny/rvm.el
ruby - 分布式事务和队列，ruby，erlang，scala - 2
我有一个涉及多台机器、消息队列和事务的问题。因此，例如用户点击网页，点击将消息发送到另一台机器，该机器将付款添加到用户的帐户。每秒可能有数千次点击。事务的所有方面都应该是容错的。我以前从未遇到过这样的事情，但一些阅读表明这是一个众所周知的问题。所以我的问题。我假设安全的方法是使用两阶段提交，但协议(protocol)是阻塞的，所以我不会获得所需的性能，我是否正确？我通常写Ruby，但似乎Redis之类的数据库和Rescue、RabbitMQ等消息队列系统对我的帮助不大——即使我实现某种两阶段提交，如果Redis崩溃，数据也会丢失，因为它本质上只是内存。所有这些让我开始关注erlang和
ruby-on-rails - 如何在发布新的 Ruby 或 Rails 版本时收到通知？ - 2
有人知道在发布新版本的Ruby和Rails时收到电子邮件的方法吗？他们有邮件列表，RubyonRails有一个推特，但我不想听到那些随之而来的喧嚣，我只想知道什么时候发布新版本，尤其是那些有安全修复的版本。最佳答案从therailsblog获取提要.http://weblog.rubyonrails.org/feed/atom.xml 关于ruby-on-rails-如何在发布新的Ruby或Rails版本时收到通知？，我们在StackOverflow上找到一个类似的问题：
【鸿蒙应用开发系列】- 获取系统设备信息以及版本API兼容调用方式 - 2
在应用开发中，有时候我们需要获取系统的设备信息，用于数据上报和行为分析。那在鸿蒙系统中，我们应该怎么去获取设备的系统信息呢，比如说获取手机的系统版本号、手机的制造商、手机型号等数据。1、获取方式这里分为两种情况，一种是设备信息的获取，一种是系统信息的获取。1.1、获取设备信息获取设备信息，鸿蒙的SDK包为我们提供了DeviceInfo类，通过该类的一些静态方法，可以获取设备信息，DeviceInfo类的包路径为：ohos.system.DeviceInfo.具体的方法如下：ModifierandTypeMethodDescriptionstatic StringgetAbiList()Obt
ruby-on-rails - 安装多个版本的 Rails 会覆盖以前的安装吗？ - 2
如果我一直输入geminstallrails使用不同版本的Rails会怎样？例如，我可以输入:geminstallrails--verson3.2.10或geminstallrails这给了我版本3.2.12。问题每次安装都会覆盖之前的吗？它会删除所有旧文件并添加我正在安装的新版本吗？或者如果我运行它两次，它会保留一些文件吗？我正在使用Ubuntu。最佳答案它将安装两个独立的gem。实际的可执行文件rails将调用最新版本。你可以覆盖它__例如，rails_3.2.10_将执行Rails3.2.10。bundler顺便说一下，如
ruby - 在 Rails 项目中测试本地版本的 gem - 2
我的Rails站点使用了一个确实不是很好的gem。每次我需要做一些新的事情时，我最终不得不花费与向实际Rails项目添加代码一样多的时间来为gem添加功能。但我不介意，我将我的Gemfile设置为指向我的gem的GitHub分支(我尝试提交PR，但维护者似乎已经下台)。问题是我真的没有找到一种合理的方法来测试我添加到gem的新东西。在railsc中测试它会特别好，但我能想到的唯一方法是a)更改~/.rvm/gems/.../foo。rb，这看起来不对或者b)升级版本，推送到Github，然后运行bundleup，这除了耗时之外显然是一场灾难，因为我不确定我所做的promise是否正
ruby-on-rails - 不兼容的库版本 : nokogiri. bundle 需要 8.0.0 或更高版本，但 libiconv.2.dylib 提供 7.0.0 版本 - 2
为了在我的mac上为一个rails项目安装mysql，我遵循了安装Homebrew软件和删除mac端口的在线建议。这是问题开始的地方。rails项目不会构建，我得到这个:[rake--prereqs]rakeaborted!dlopen(/Users/Parker/.rvm/gems/ruby-1.9.3-p448/gems/nokogiri-1.6.0/lib/nokogiri/nokogiri.bundle,9):Librarynotloaded:/opt/local/lib/libiconv.2.dylibReferencedfrom:/Users/Parker/.rvm/gem

Spark算子-Scala版本 头歌答案