Azure 上的 WordCount 使用 hadoop 和 spark

Question

我必须在 Azure 中的集群 Hdinsight 上运行一个简单的字数统计。我用hadoop和spark创建了一个集群，我已经有了带有代码的jar文件，问题是我不知道如何设置集群和正确的代码行来在Azure上启动spark，我想尝试不同的节点组合（workers，2-4-8），看看程序如何扩展。

每次我在 yarn-client 模式下使用 spark-submit 启动应用程序时，它都可以工作，但总是在 3 分钟左右为 1gb 输入文本文件使用 2 个执行器和 1 个核心，如果我设置更多执行器和更多核心，他也会使用设置，但他不使用它，所以我认为问题出在 RDD 上，它不会以正确的模式拆分输入文件，因为它只创建了 2 个从 2 个工作节点开始的任务，而其他节点保持非活动状态.

它是用 sbt 包创建的 jar 文件。

启动 Spark 的命令：

spark-submit --class "SimpleApp" --master yarn-client --num-executors 2 simpleapp_2.10-1.0.jar

字数代码：

import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import java.io._
import org.apache.hadoop.fs
import org.apache.spark.rdd.RDD


object SimpleApp {
def main(args: Array[String]){
//settingsparkcontext
val conf = new SparkConf().setAppName("SimpleApp")
val sc = new SparkContext(conf)
//settingthewordtosearch
val word = "word"
//settingtime
val now = System.nanoTime
//settingtheinputfile
val input = sc.textFile("wasb://xxx@storage.blob.core.windows.net/dizionario1gb.txt")

//wordlookup
val splittedLines = input.map(line=>line.split(""))
val find = System.nanoTime
val tot = splittedLines.map(x => x.equals(word)).count()
val w=(System.nanoTime-find)/1000000
val rw=(System.nanoTime-now)/1000000
//reportingtheresultofexecutioninatxtfile
val writer = new FileWriter("D:\\Users\\user\\Desktop\\File\\output.txt",true)
try {
writer.write("Word found "+tot+" time total "+rw+" mstimesearch "+w+" time read "+(rw-w)+"\n")
}
finally writer.close()
//terminatingthesparkserver
sc.stop()
}}              

Answer 1

并行度

“除非您为每个操作设置足够高的并行级别，否则集群将不会被充分利用。Spark 会根据文件大小自动设置要在每个文件上运行的“映射”任务的数量（尽管您可以通过可选的参数到 SparkContext.textFile 等）您可以将并行度级别作为第二个参数传递（请参阅 spark.PairRDDFunctions 文档），或设置配置属性 spark.default.parallelism 以更改默认值。一般来说，我们建议 2-集群中每个 CPU 核心 3 个任务。”

来源： https://spark.apache.org/docs/1.3.1/tuning.html