在 Scala 中设计和并行化 Spark 应用程序的最佳方法 [关闭]

Question

我正在使用 Scala 开发 Spark 应用程序，并且想知道将其并行化并在 Hadoop 集群上运行的最佳方法。我的代码将从 HDFS 文件中读取每一行并对其进行解析并生成多个记录（对于每一行），我将这些记录存储为案例类。我已经在 getElem() 方法中编写了完整的逻辑并按预期工作。

现在，我想计算所有输入记录的逻辑并将响应存储到 HDFS 位置。

请让我知道我将如何处理 spark 并合并为输入生成的所有相应输出记录并写入 HDFS。

        object testing extends Serializable {
      var recordArray=Array[Record]();
       def main(args:Array[String])
      {

      val conf = new SparkConf().setAppName("jsonParsing").setMaster("local")
      val sc = new SparkContext(conf)
      val sqlContext= new SQLContext(sc)

      val input=sc.textFile("hdfs://loc/data.txt")
     // input.collect().foreach(println)
      input.map(data=>getElem(parse(data,false),sc,sqlContext))

    }
          //method definition
    def getElem(json:JValue)={

         // Parses the json and creates array of datasets for each input record and stores the data in case class
  val x= Record("xxxx","xxxx","xxxx","xxxx","xxxx","xxxx","xxxx","xxxx","xxxx","xxxx")
             }
  case class Record(summary_key: String, key: String,array_name_position:Int,Parent_Level_1:String,Parent_level_2:String,Parent_Level_3:String,Parent_level_4:String,Parent_level_5:String,
        param_name_position:Integer,Array_name:String,paramname:String,paramvalue:String)
           }

Answer 1

你说你有工作 case class 并且你有 parsed 来自 hdfs 的 input text file 并转换为 record case class。
然后使用.toDF() 或.toDS() 转换为dataframe 或dataset 将很容易
现在剩下的任务就是将case class rdd或dataframe或dataset写入hdfs
将案例类rdd写入hdfs：
只需调用.saveAsTextFile() api，您就可以提供output directory 的路径，就像您从hdfs 读取一样
将数据帧或数据集写入hdfs：
你可以使用databricks api .write.format("com.databricks.spark.csv").save() 在那里你可以给输出hdfs dir name

我希望这有帮助

注意：我猜你是通过ide 运行代码。你应该学会spark-submit更多信息可以在here找到