怎么實踐Spark，針對這個問題，這篇文章詳細介紹了相對應的分析和解答，希望可以幫助更多想解決這個問題的小伙伴找到更簡單易行的方法。

創新互聯建站是一家以重慶網站建設公司、網頁設計、品牌設計、軟件運維、營銷推廣、小程序App開發等移動開發為一體互聯網公司。已累計為成都花箱等眾行業中小客戶提供優質的互聯網建站和軟件開發服務。

Spark小試牛刀

隨著項目的運營，收集了很多的用戶數據。最近業務上想做些社交圖譜相關的產品，但因為數據很多、很雜，傳統的數據庫查詢已經滿足不了業務的需求。 試著用Spark來做，權當練練手了。

安裝Spark

因為有Scala的開發經驗，所以就不用官方提供的二進制包了，自編譯scala 2.11版本。

下載Spark：http://ftp.cuhk.edu.hk/pub/packages/apache.org/spark/spark-1.5.0/spark-1.5.0.tgz

tar zxf spark-1.5.0.tgz
cd spark-1.5.0
./dev/change-scala-version.sh 2.11
mvn -Pyarn -Phadoop-2.6 -Dscala-2.11 -DskipTests clean package

以上命令完成Spark基于scala 2.11版本的編譯。可以運行自帶的一個示例程序來驗證安裝是否成功。

./bin/run-example SparkPi

編寫Standalone application

使用sbt來構建一個可提交的簡單Spark程序，功能是計算每個用戶加入的群組，并把結果保存下來。project/Build.scala配置文件如下：

import _root_.sbt.Keys._
import _root_.sbt._
import sbtassembly.AssemblyKeys._

object Build extends Build {

  override lazy val settings = super.settings :+ {
    shellPrompt := (s => Project.extract(s).currentProject.id + " > ")
  }

  lazy val root = Project("spark-MongoDB", file("."))
    .settings(
      scalaVersion := "2.11.7",
      assemblyJarName in assembly := "spark-mongodb.jar",
      assemblyOption in assembly := (assemblyOption in assembly).value.copy(includeScala = false),
      libraryDependencies ++= Seq(
        "org.apache.spark" %% "spark-core" % verSpark % "scopeProvidedTest,
        "org.mongodb.mongo-hadoop" % "mongo-hadoop-core" % "1.4.0" excludeAll(
            ExclusionRule(organization = "javax.servlet"), 
            ExclusionRule(organization = "commons-beanutils"), 
            ExclusionRule(organization = "org.apache.hadoop")))
    )
  
  private val scopeProvidedTest = "provided,test"
  private val verSpark = "1.5.0"
}

數據存儲在MongoDB數據庫中，所以我們還需要使用mongo-hadoop連接器來訪問MongoDB數據庫。

示例程序

示例程序非常的簡單，把數據從數據庫里全部讀出，使用map來把每條記錄里用戶ID對應加入的群組ID轉換成一個Set，再使用 reduceByKey來把相同用戶ID的set合并到一起，存入數據庫即可。

import com.mongodb.BasicDBObject
import com.mongodb.hadoop.{MongoInputFormat, MongoOutputFormat}
import org.apache.hadoop.conf.Configuration
import org.apache.spark.{SparkConf, SparkContext}
import org.bson.BSONObject

import scala.collection.JavaConverters._

object QQGroup {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setAppName("QQGroup")
    val sc = new SparkContext(sparkConf)

    val inputConfig = new Configuration()
    inputConfig.set("mongo.input.uri", "mongodb://192.168.31.121:27017/db.userGroup")
    inputConfig.set("mongo.input.fields", """{"userId":1, "groupId":1, "_id":0}""")
    inputConfig.set("mongo.input.noTimeout", "true")

    val documentRDD = sc.newAPIHadoopRDD(
      inputConfig,
      classOf[MongoInputFormat],
      classOf[Object],
      classOf[BSONObject])

    val userRDD = documentRDD.map { case (_, doc) =>
      (getValue(doc, "userId"), getValue(doc, "groupId"))
    }.reduceByKey(_ ++ _)

    val resultRDD = userRDD.map { case (userId, groupIds) =>
      val o = new BasicDBObject()
      o.put("groupIds", groupIds.asJava)
      userId -> o
    }

    val outputConfig = new Configuration()
    outputConfig.set("mongo.output.uri", "mongodb://192.168.31.121:27017/db_result.userGroup")

    resultRDD.saveAsNewAPIHadoopFile(
      "file://this-is-completely-unused",
      classOf[Object],
      classOf[BSONObject],
      classOf[MongoOutputFormat[Object, BSONObject]],
      outputConfig)
  }

  def getValue(dbo: BSONObject, key: String) = {
    val value = dbo.get(key)
    if (value eq null) "" else value.asInstanceOf[String]
  }
}

MongoDB官方提供了Hadoop連接器，Spark可以使用mongo-hadoop連接器來讀、寫MongoDB數據庫。 主要的輸入配置薦有：

• mongo.input.uri: MongoDB的連接URI

• mongo.input.fields: 指定返回哪些數據，與db.query里的第2個參數功能一樣

• mongo.input.query: MongoDB的查詢參數

相應的MongoDB也提供了一系列的輸出參數，如：

• mongo.output.uri: MongoDB的連接URI

sc.newAPIHadoopRDD()方法有4個參數，分別為：配置、輸入格式化類、待映射數據主鍵類型、待映射數據類型。

主要的操作代碼：

    val userRDD = documentRDD.map { case (_, doc) =>
      (getValue(doc, "userId"), Set(getValue(doc, "groupId")))
    }.reduceByKey(_ ++ _)

    val resultRDD = userRDD.map { case (userId, groupIds) =>
      val o = new BasicDBObject()
      o.put("groupIds", groupIds.asJava)
      userId -> o
    }

先使用map方法獲取userId和groupId，并把groupId轉換為一個Set。

在把數據轉換成Tuple2，就是一個KV的形式以后，我們就可以調用一系列的轉換方法來對RDD進行操作，這里使用reduceByKey方法來將同一個userId的所以value都合并在一起。這樣我們就有了所有用戶對應加入的群組 的一個RDD集了。

（RDD上有兩種類型的操作。一種是“變換”，它只是描述了待進行的操作指令，并不會觸發實際的計算；另一種是“動作”， 它將觸發實際的計算動作，這時候系統才會實際的從數據源讀入數據，操作內存，保存數據等）

最后使用resultRDD.saveAsNewAPIHadoopFile()方法來把計算結果存入MongoDB，這里的一個參數：用于指定 HDFS的存儲位置并不會使用到，因為mongo-hadoop將會使用mongo.output.uri指定的存儲URI連接地址來保存數據。

關于怎么實踐Spark問題的解答就分享到這里了，希望以上內容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關注創新互聯行業資訊頻道了解更多相關知識。

文章標題：怎么實踐Spark
當前路徑：http://m.2m8n56k.cn/article46/jdcheg.html

成都網站建設公司_創新互聯，為您提供ChatGPT、營銷型網站建設、網站排名、移動網站建設、電子商務、網站收錄

廣告

聲明：本網站發布的內容（圖片、視頻和文字）以用戶投稿、用戶轉載內容為主，如果涉及侵權請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網站立場，如需處理請聯系客服。電話：028-86922220；郵箱：631063699@qq.com。內容未經允許不得轉載，或轉載時需注明來源： 創新互聯