我有一个 spark 作业，它从 hdfs 获取一个包含 8 条记录的文件，进行简单的聚合并将其保存回 hdfs。我注意到执行此操作时有数百个任务。
我也不确定为什么会有多个作业？我认为工作更像是一个 Action 发生的时候。我可以推测原因——但我的理解是，在这段代码中，它应该是一项工作，应该分解成多个阶段，而不是多项工作。为什么不把它分解成阶段，它怎么分解成工作？
就 200 多个任务而言，由于数据量和节点数量微乎其微，因此当只有一个聚合和一对时，每行数据有 25 个任务是没有意义的的过滤器。为什么每个原子操作的每个分区不只有一个任务？
这是相关的 scala 代码 -
import org.apache.spark.sql._
import org.apache.spark.sql.types._
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
object TestProj {object TestProj {
def main(args: Array[String]) {
/* set the application name in the SparkConf object */
val appConf = new SparkConf().setAppName("Test Proj")
/* env settings that I don't need to set in REPL*/
val sc = new SparkContext(appConf)
val sqlContext = new SQLContext(sc)
import sqlContext.implicits._
val rdd1 = sc.textFile("hdfs://node002:8020/flat_files/miscellaneous/ex.txt")
/*the below rdd will have schema defined in Record class*/
val rddCase = sc.textFile("hdfs://node002:8020/flat_files/miscellaneous/ex.txt")
.map(x=>x.split(" ")) //file record into array of strings based spaces
.map(x=>Record(
x(0).toInt,
x(1).asInstanceOf[String],
x(2).asInstanceOf[String],
x(3).toInt))
/* the below dataframe groups on first letter of first name and counts it*/
val aggDF = rddCase.toDF()
.groupBy($"firstName".substr(1,1).alias("firstLetter"))
.count
.orderBy($"firstLetter")
/* save to hdfs*/
aggDF.write.format("parquet").mode("append").save("/raw/miscellaneous/ex_out_agg")
}
case class Record(id: Int
, firstName: String
, lastName: String
, quantity:Int)
}
下面是点击应用后的截图
以下是查看 id 0 的特定“作业”时显示的阶段
下面是点击超过200个任务的舞台时的第一部分画面
这是舞台内屏幕的第二部分
下面是点击“executors”选项卡后的样子
根据要求，这是作业 ID 1 的阶段
以下是作业 ID 1 中包含 200 个任务的阶段的详细信息
最佳答案
这是一个经典的 Spark 问题。
用于读取的两个任务(第二个图中的 Stage Id 0)是 defaultMinPartitions 设置为 2。您可以通过读取 REPL sc 中的值来获取此参数.defaultMinPartitions。它也应该在“环境”选项卡下的 Spark UI 中可见。
你可以看看code从 GitHub 上看到这正是正在发生的事情。如果您希望在读取时使用更多分区，只需将其添加为参数即可，例如 sc.textFile("a.txt", 20)。
现在有趣的部分来自第二阶段的 200 个分区(第二个图中的阶段 Id 1)。好吧，每次有一个 shuffle，Spark 需要决定这个 shuffle 的 RDD 有多少个分区。可以想象，默认值为 200。
您可以使用以下方式更改它:
sqlContext.setConf("spark.sql.shuffle.partitions", "4”)
如果您使用此配置运行代码，您将看到 200 个分区将不再存在。如何设置这个参数是一门艺术。也许选择 2 倍于您拥有的核心数(或其他)。
我认为 Spark 2.0 有一种方法可以自动推断 shuffle RDD 的最佳分区数。期待那个!
最后，您获得的作业数量与生成的优化 Dataframe 代码产生的RDD 操作 数量有关。如果您阅读 Spark 规范，它会说每个 RDD 操作都会触发一个作业。当您的操作涉及 Dataframe 或 SparkSQL 时，Catalyst 优化器将找出执行计划并生成一些基于 RDD 的代码来执行它。很难确切地说出为什么它在您的案例中使用了两个动作。您可能需要查看优化的查询计划以准确了解正在执行的操作。
关于scala - 为什么我的 Spark 作业中有这么多任务？默认获取 200 个任务，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/37758647/
类classAprivatedeffooputs:fooendpublicdefbarputs:barendprivatedefzimputs:zimendprotecteddefdibputs:dibendendA的实例a=A.new测试a.foorescueputs:faila.barrescueputs:faila.zimrescueputs:faila.dibrescueputs:faila.gazrescueputs:fail测试输出failbarfailfailfail.发送测试[:foo,:bar,:zim,:dib,:gaz].each{|m|a.send(m)resc
我试图在一个项目中使用rake，如果我把所有东西都放到Rakefile中，它会很大并且很难读取/找到东西，所以我试着将每个命名空间放在lib/rake中它自己的文件中，我添加了这个到我的rake文件的顶部:Dir['#{File.dirname(__FILE__)}/lib/rake/*.rake'].map{|f|requiref}它加载文件没问题，但没有任务。我现在只有一个.rake文件作为测试，名为“servers.rake”，它看起来像这样:namespace:serverdotask:testdoputs"test"endend所以当我运行rakeserver:testid时
我有一个模型:classItem项目有一个属性“商店”基于存储的值，我希望Item对象对特定方法具有不同的行为。Rails中是否有针对此的通用设计模式？如果方法中没有大的if-else语句，这是如何干净利落地完成的？ 最佳答案 通常通过Single-TableInheritance. 关于ruby-on-rails-Rails-子类化模型的设计模式是什么？，我们在StackOverflow上找到一个类似的问题： https://stackoverflow.co
我正在使用的第三方API的文档状态:"[O]urAPIonlyacceptspaddedBase64encodedstrings."什么是“填充的Base64编码字符串”以及如何在Ruby中生成它们。下面的代码是我第一次尝试创建转换为Base64的JSON格式数据。xa=Base64.encode64(a.to_json) 最佳答案 他们说的padding其实就是Base64本身的一部分。它是末尾的“=”和“==”。Base64将3个字节的数据包编码为4个编码字符。所以如果你的输入数据有长度n和n%3=1=>"=="末尾用于填充n%
我主要使用Ruby来执行此操作，但到目前为止我的攻击计划如下:使用gemsrdf、rdf-rdfa和rdf-microdata或mida来解析给定任何URI的数据。我认为最好映射到像schema.org这样的统一模式，例如使用这个yaml文件，它试图描述数据词汇表和opengraph到schema.org之间的转换:#SchemaXtoschema.orgconversion#data-vocabularyDV:name:namestreet-address:streetAddressregion:addressRegionlocality:addressLocalityphoto:i
为什么4.1%2返回0.0999999999999996？但是4.2%2==0.2。 最佳答案 参见此处:WhatEveryProgrammerShouldKnowAboutFloating-PointArithmetic实数是无限的。计算机使用的位数有限(今天是32位、64位)。因此计算机进行的浮点运算不能代表所有的实数。0.1是这些数字之一。请注意，这不是与Ruby相关的问题，而是与所有编程语言相关的问题，因为它来自计算机表示实数的方式。 关于ruby-为什么4.1%2使用Ruby返
如何使用RSpec::Core::RakeTask初始化RSpecRake任务？require'rspec/core/rake_task'RSpec::Core::RakeTask.newdo|t|#whatdoIputinhere?endInitialize函数记录在http://rubydoc.info/github/rspec/rspec-core/RSpec/Core/RakeTask#initialize-instance_method没有很好的记录；它只是说:-(RakeTask)initialize(*args,&task_block)AnewinstanceofRake
它不等于主线程的binding，这个toplevel作用域是什么？此作用域与主线程中的binding有何不同？>ruby-e'putsTOPLEVEL_BINDING===binding'false 最佳答案 事实是，TOPLEVEL_BINDING始终引用Binding的预定义全局实例，而Kernel#binding创建的新实例>Binding每次封装当前执行上下文。在顶层，它们都包含相同的绑定(bind)，但它们不是同一个对象，您无法使用==或===测试它们的绑定(bind)相等性。putsTOPLEVEL_BINDINGput
我可以得到Infinity和NaNn=9.0/0#=>Infinityn.class#=>Floatm=0/0.0#=>NaNm.class#=>Float但是当我想直接访问Infinity或NaN时:Infinity#=>uninitializedconstantInfinity(NameError)NaN#=>uninitializedconstantNaN(NameError)什么是Infinity和NaN？它们是对象、关键字还是其他东西？ 最佳答案 您看到打印为Infinity和NaN的只是Float类的两个特殊实例的字符串
如果您尝试在Ruby中的nil对象上调用方法，则会出现NoMethodError异常并显示消息:"undefinedmethod‘...’fornil:NilClass"然而，有一个tryRails中的方法，如果它被发送到一个nil对象，它只返回nil:require'rubygems'require'active_support/all'nil.try(:nonexisting_method)#noNoMethodErrorexceptionanymore那么try如何在内部工作以防止该异常？ 最佳答案 像Ruby中的所有其他对象