TMDB電影數(shù)據(jù)分析（spark-scala版，pyspark版本）

基于kaggle的TMDB電影數(shù)據(jù)集的數(shù)據(jù)分析，該數(shù)據(jù)集包含大約 5000 部電影的相關(guān)數(shù)據(jù)，TMDB數(shù)據(jù)下載。分析電影關(guān)鍵詞的分布、電影投入以及收益評價等之間的關(guān)系，并使用Python web進(jìn)行可視化。

一.環(huán)境要求（僅供參考）

centos7、hadoop、spark、bottle（一種基于Python的web框架）。

二.數(shù)據(jù)預(yù)處理

tmdb_5000_movies.csv 數(shù)據(jù)包含以下字段：
字段名稱 解釋 例子
budget 預(yù)算 10000000
genres 體裁 “[{“”id””: 18, “”name””: “”Drama””}]”
homepage 主頁 “”
id id 268238
keywords 關(guān)鍵詞 “[{“”id””: 14636, “”name””: “”india””}]”
original_language 原始語言 en
original_title 原標(biāo)題 The Second Best Exotic Marigold Hotel
overview 概覽 As the Best Exotic Marigold Hotel …
popularity 流行度 17.592299
production_companies 生產(chǎn)公司 “[{“”name””: “”Fox Searchlight Pictures””, “”id””: 43}, …]”
production_countries 生產(chǎn)國家 “[{“”iso31661″”: “”GB””, “”name””: “”United Kingdom””}, …]”
release_date 發(fā)行日期 2015-02-26
revenue 盈收 85978266
runtime 片長 122
spoken_languages 語言 “[{“”iso6391″”: “”en””, “”name””: “”English””}]”
status 狀態(tài) Released
tagline 宣傳語 “”
title 標(biāo)題 The Second Best Exotic Marigold Hotel
vote_average 平均分 6.3
vote_count 投票人數(shù) 272

• 對數(shù)據(jù)去除標(biāo)題行。數(shù)據(jù)中某些字段包含 json 數(shù)據(jù)，直接使用 DataFrame 進(jìn)行讀取會出現(xiàn)分割錯誤，如果要創(chuàng)建 DataFrame，先讀取文件生成 RDD，再將 RDD 轉(zhuǎn)為 DataFrame。
• 為了便于處理上傳數(shù)據(jù)至hdfs文件系統(tǒng)：

hdfs dfs -put tmdb_5000_movies.csv

三、使用 Spark 將數(shù)據(jù)轉(zhuǎn)為 DataFrame

為了創(chuàng)建 DataFrame，首先需要將 HDFS 上的數(shù)據(jù)加載成 RDD，再將 RDD 轉(zhuǎn)化為 DataFrame。下面代碼段完成從文件到 RDD 再到 DataFrame 的轉(zhuǎn)化：

// 創(chuàng)建sparksession import org.apache.spark.sql.SparkSession val spark=SparkSession.builder().getOrCreate() import spark.implicits._ // 使用編程方式定義RDD模式 import org.apache.spark.sql.types._ import org.apache.spark.sql.Row // 定義一個模式字符串 val schemaString="budget,genres,homepage,id,keywords,original_language,original_title,overview,popularity,production_companies,production_countries,release_date,revenue,runtime,spoken_languages,status,tagline,title,vote_average,vote_count" // 根據(jù)字符串生成模式 val fields=schemaString.split(",").map(fieldName => StructField(fieldName,StringType,nullable=true)) val schema=StructType(fields) val path = "hdfs://192.168.1.30:9000/user/root/tmdb_5000_movies.csv"// 由于tmdb的csv數(shù)據(jù)某些字段中包含 val mdf=spark.read.format("com.databricks.spark.csv").schema(schema).option("inferSchema", value = false).option("header", value = true).option("nullValue", "\\N").option("escape", "\"") // 設(shè)置用于在已引用的值內(nèi)轉(zhuǎn)義引號的單個字符。詳情見 spark 讀取 csv 官網(wǎng)介紹 https://spark.apache.org/docs/2.0.2/api/java/org/apache/spark/sql/DataFrameReader.html#option(java.lang.String,%20boolean).option("quoteAll","true").option("sep", ",").csv(path) mdf.select("genres").show(2,false)

四、使用 Spark 進(jìn)行數(shù)據(jù)分析

Spark 處理得到的 DataFrame mdf 進(jìn)行數(shù)據(jù)分析，首先對數(shù)據(jù)中的主要字段單獨(dú)進(jìn)行分析（概覽小節(jié)），然后再分析不同字段間的關(guān)系（關(guān)系小節(jié)）。為了方便進(jìn)行數(shù)據(jù)可視化，每個不同的分析，都將分析結(jié)果導(dǎo)出為 json 文件存儲到static目錄下，由 web 頁面讀取并進(jìn)行可視化。

1.概覽

這個部分對數(shù)據(jù)進(jìn)行整體的分析。

1.1.TMDb 電影中的體裁分布

從上面的數(shù)據(jù)字典描述可以看出，電影的體裁字段是一個 json 格式的數(shù)據(jù)，因此，為了統(tǒng)計不同體裁的電影的數(shù)量，需要首先解析 json 數(shù)據(jù)，從中取出每個電影對應(yīng)的體裁數(shù)組，然后使用詞頻統(tǒng)計的方法統(tǒng)計不同體裁出現(xiàn)的頻率，即可得到電影的體裁分布。
首先實(shí)現(xiàn)一個函數(shù) countByJson(field) ，該函數(shù)實(shí)現(xiàn)解析 json 格式字段從中提取出 name 并進(jìn)行詞頻統(tǒng)計的功能：

// （1）TMDb 電影中的體裁分布 import org.apache.spark.sql.functions._ import org.apache.spark.rdd.RDD import org.apache.spark.sql.DataFrame import org.apache.spark.sql.Rowimport org.json4s._ import org.json4s.JsonDSL._ import org.json4s.jackson.JsonMethods._import java.io.PrintWriter //這行是Scala解釋器執(zhí)行上面語句后返回的結(jié)果 // // 尋找tmdb中最常見的10中預(yù)算數(shù)，對電影預(yù)算頻率進(jìn)行統(tǒng)計 // // 使用sql語句分析 // 或者使用spark的轉(zhuǎn)換函數(shù)進(jìn)行分析 def countByJson(field:String):org.apache.spark.rdd.RDD[(String,Int)] ={val jsonSchema =ArrayType(new StructType().add("id", IntegerType).add("name",StringType)) mdf.select(mdf.col(field)).filter(mdf.col(field).isNotNull) // 此處是單條中包含多個數(shù)據(jù)的json，按照jsonSchema的格式進(jìn)行解析，并生成多條單個數(shù)據(jù)，explode是將數(shù)組組生成為列。.select(explode(from_json(mdf.col(field), jsonSchema)).as(field)) // 解決$"genres.name"的變量問題.select(field.concat(".name")).rdd.map(name=>(name.toString(),1)).repartition(1).reduceByKey((x,y) => x + y) }

該函數(shù)返回一個 RDD，整個過程如下圖所示。

基于這個函數(shù)實(shí)現(xiàn) countByGenres 用來生成不同體裁的電影數(shù)統(tǒng)計結(jié)果，接著，使用下面代碼進(jìn)行數(shù)據(jù)導(dǎo)出至 genres.json 方便之后進(jìn)行可視化：

def countByGenres():String={val genresRDD=countByJson("genres")val jsonString =genresRDD.collect().toList.map { case(genre,count) =>(("genre" ->genre.replace("[","").replace("]",""))~("count" ->count))}val mdfJson=compact(render(jsonString))mdfJson } def save(str:String,path:String,hdfspath:String):Unit={ // 寫入本地文件val out = new PrintWriter(path)out.println(str)out.close() // 寫入hdfs文件系統(tǒng)val str_rdd=spark.sparkContext.parallelize(List(str))str_rdd.saveAsTextFile(hdfspath) } val str=countByGenres() println(str) val path="/home/chenbengang/ziyu_bigdata/quick_learn_spark/movie_genres_word_count.txt" val hdfspath="hdfs://192.168.1.30:9000/user/root/movie_genres_word_count.txt" save(str,path,hdfspath)

1.2. 前 100 個常見關(guān)鍵詞

該項分析電影關(guān)鍵詞中出現(xiàn)頻率最高的前一百個。由于關(guān)鍵詞字段也是 json 格式數(shù)據(jù)，因此調(diào)用 countByJson 進(jìn)行頻率統(tǒng)計，同時對于統(tǒng)計結(jié)果進(jìn)行降序排序并取前 100 項即可：

// 2. 前 100 個常見關(guān)鍵詞 def countByKeywords():String={ // 對rdd排序val keywordsRDD=countByJson("keywords").sortBy(x=>(-x._2))val jsonString =keywordsRDD.take(100).toList.map { case(keywords,count) =>(("keywords" ->keywords.replace("[","").replace("]",""))~("count" ->count))}val mdfJson=compact(render(jsonString))mdfJson } val str=countByKeywords() println(str) val path="/home/chenbengang/ziyu_bigdata/quick_learn_spark/movie_keywords_word_count.txt" val hdfspath="hdfs://192.168.1.30:9000/user/root/movie_keywords_word_count.txt" save(str,path,hdfspath)

1.3. TMDb 中最常見的 10 種預(yù)算數(shù)

這一項探究電影常見的預(yù)算數(shù)是多少，因此需要對電影預(yù)算進(jìn)行頻率統(tǒng)計。首先，需要對預(yù)算字段進(jìn)行過濾，去除預(yù)算為 0 的項目，然后根據(jù)預(yù)算聚合并計數(shù)，接著根據(jù)計數(shù)進(jìn)行排序，并將結(jié)果導(dǎo)出為 json 字符串，為了統(tǒng)一輸出，這里將 json 字符串轉(zhuǎn)為 python 對象，最后取前 10 項作為最終的結(jié)果。

// 3. TMDb 中最常見的 10 種預(yù)算數(shù) def countByBudget(order:String,ascending:Boolean):Array[String]={if (ascending){mdf.filter(!$"budget".equalTo(0)).groupBy("budget").count().orderBy(order).toJSON.take(10)}else{mdf.filter(!$"budget".equalTo(0)).groupBy("budget").count().orderBy(desc(order)).toJSON.take(10)} } val budgetTop10Arr=countByBudget("count",false)val movie_budgetTop10 = new StringBuilder movie_budgetTop10 ++= "[" for (v <- budgetTop10Arr){movie_budgetTop10 ++= v } movie_budgetTop10 ++= "]" println(movie_budgetTop10.toString) val path="/home/chenbengang/ziyu_bigdata/quick_learn_spark/movie_budgetTop10.txt" val hdfspath="hdfs://192.168.1.30:9000/user/root/movie_budgetTop10.txt" save(movie_budgetTop10.toString,path,hdfspath)

1.4. TMDb 中最常見電影時長 (只展示電影數(shù)大于 100 的時長)

這一項統(tǒng)計 TMDb 中最常見的電影時長，首先，需要過濾時長為 0 的電影，然后根據(jù)時長字段聚合并計數(shù)，接著過濾掉出現(xiàn)頻率小于 100 的時長 （這一步是為了方便可視化，避免過多冗余信息）得到最終的結(jié)果。

// 4. TMDb 中最常見電影時長 (只展示電影數(shù)大于 100 的時長) def distrbutionOfRuntime(order:String,ascending:Boolean):Array[String]={ // 后一個filter之前是dataset有兩列一列runtime，一列為countmdf.filter(!$"runtime".equalTo(0)).groupBy("runtime").count().filter("count>=100").toJSON.collect() } val runtimeOfCountOver100Arr=distrbutionOfRuntime("count",false) val movie_runtimeOfCountOver100 = new StringBuilder movie_runtimeOfCountOver100 ++= "[" for (v <- runtimeOfCountOver100Arr){movie_runtimeOfCountOver100 ++= v } movie_runtimeOfCountOver100 ++= "]" println(movie_runtimeOfCountOver100.toString) val path="/home/chenbengang/ziyu_bigdata/quick_learn_spark/movie_runtimeOfCountOver100.txt" val hdfspath="hdfs://192.168.1.30:9000/user/root/movie_runtimeOfCountOver100.txt" save(movie_runtimeOfCountOver100.toString,path,hdfspath)

1.5. 生產(chǎn)電影最多的 10 大公司

這一項統(tǒng)計電影產(chǎn)出最多的 10 個公司，同樣使用 countByJson 對 JSON 數(shù)據(jù)進(jìn)行頻率統(tǒng)計，然后進(jìn)行降序排列取前 10 項即可。

// 5. 生產(chǎn)電影最多的 10 大公司 def countByCompanies():String={val production_companiesRDD=countByJson("production_companies").sortBy(x=>(-x._2))val jsonString =production_companiesRDD.take(10).toList.map { case(company,count) =>(("company" ->company.replace("[","").replace("]",""))~("count" ->count))}val mdfJson=compact(render(jsonString))mdfJson } val movie_countByCompanies=countByCompanies() println(movie_countByCompanies) val path="/home/chenbengang/ziyu_bigdata/quick_learn_spark/movie_countByCompanies.txt" val hdfspath="hdfs://192.168.1.30:9000/user/root/movie_countByCompanies.txt" save(str,path,hdfspath)

1.6. TMDb 中的 10 大電影語言

該項統(tǒng)計 TMDb 中出現(xiàn)最多的語言，與前面類似，該字段也是 JSON 數(shù)據(jù)，因此首先對每個項目進(jìn)行詞頻統(tǒng)計，然后過濾掉語言為空的項目，最后排序取前十即可。

// 6. TMDb 中的 10 大電影語言 def countByLanguageRDD():String={val countByLanguageRDD=countByJson("spoken_languages").sortBy(x=>(-x._2))val jsonString =countByLanguageRDD.take(10).toList.map { case(language,count) =>(("language" ->language.replace("[","").replace("]",""))~("count" ->count))}val mdfJson=compact(render(jsonString))mdfJson } val movie_countByLanguage=countByLanguageRDD() println(movie_countByLanguage) val path="/home/chenbengang/ziyu_bigdata/quick_learn_spark/movie_countByLanguage.txt" val hdfspath="hdfs://192.168.1.30:9000/user/root/movie_countByLanguage.txt" save(movie_countByLanguage,path,hdfspath)

2.關(guān)系

這個部分考慮數(shù)據(jù)之間的關(guān)系。

2.1.預(yù)算與評價的關(guān)系

這部分考慮預(yù)算與評價之間的關(guān)系，因此對于每個電影，需要導(dǎo)出如下的數(shù)據(jù)：

• [電影標(biāo)題，預(yù)算，評價]

// 2.關(guān)系 // 這個部分考慮數(shù)據(jù)之間的關(guān)系。 // 1. 預(yù)算與評價的關(guān)系 def budgetVote():Array[String]={mdf.select($"title",$"budget",$"vote_average").filter(!$"budget".equalTo(0)).filter(mdf.col("vote_count")>100).toJSON.collect() } val budgetVoteArr=budgetVote() // println(budgetVoteArr.length) val budgetVoteSB = new StringBuilder budgetVoteSB ++= "[" for (v <- budgetVoteArr){budgetVoteSB ++= v } budgetVoteSB ++= "]" println(budgetVoteSB.toString) val path="/home/chenbengang/ziyu_bigdata/quick_learn_spark/movie_budgetVote.txt" val hdfspath="hdfs://192.168.1.30:9000/user/root/movie_budgetVote.txt" save(budgetVoteSB.toString,path,hdfspath)

2.2. 發(fā)行時間與評價的關(guān)系

這部分考慮發(fā)行時間與評價之間的關(guān)系，因此對于每個電影，需要導(dǎo)出如下的數(shù)據(jù)：

• [電影標(biāo)題，發(fā)行時間，評價]

// 2. 發(fā)行時間與評價的關(guān)系 def dateVote():Array[String]={mdf.select($"release_date",$"vote_average",$"title") .filter(mdf.col("release_date").isNotNull).filter(mdf.col("vote_count")>100).toJSON.collect() } val dateVoteArr=dateVote() // println(budgetVoteArr.length) val dateVoteSB = new StringBuilder dateVoteSB ++= "[" for (v <- dateVoteArr){dateVoteSB ++= v } dateVoteSB ++= "]" println(dateVoteSB.toString) val path="/home/chenbengang/ziyu_bigdata/quick_learn_spark/movie_dateVote.txt" val hdfspath="hdfs://192.168.1.30:9000/user/root/movie_dateVote.txt" save(dateVoteSB.toString,path,hdfspath)

2.3. 流行度和評價的關(guān)系

這部分考慮流行度與評價之間的關(guān)系，因此對于每個電影，需要導(dǎo)出如下的數(shù)據(jù)：

• [電影標(biāo)題，流行度，評價]

// 3. 流行度和評價的關(guān)系 def popVote():Array[String]={mdf.select($"title",$"popularity",$"vote_average").filter(!$"popularity".equalTo(0)).filter(mdf.col("vote_count")>100).toJSON.collect() } val popVoteArr=popVote() println(budgetVoteArr.length) val popVoteSB = new StringBuilder popVoteSB ++= "[" for (v <- popVoteArr){popVoteSB ++= v } popVoteSB ++= "]" println(popVoteSB.toString) val path="/home/chenbengang/ziyu_bigdata/quick_learn_spark/movie_popVote.txt" val hdfspath="hdfs://192.168.1.30:9000/user/root/movie_popVote.txt" save(popVoteSB.toString,path,hdfspath)

2.4. 公司生產(chǎn)的電影平均分和數(shù)量的關(guān)系

這部分計算每個公司生產(chǎn)的電影數(shù)量及這些電影的平均分分布。首先，需要對數(shù)據(jù)進(jìn)行過濾，去掉生產(chǎn)公司字段為空和評價人數(shù)小于 100 的電影，然后對于每一條記錄，得到一條如下形式的記錄：

• [公司名，(評分，1)]
  

// 4. 公司生產(chǎn)的電影平均分和數(shù)量的關(guān)系 def moviesVote():String={val jsonSchema =ArrayType(new StructType().add("id", IntegerType).add("name",StringType)) val jsonString=mdf.filter(mdf.col("production_companies").isNotNull).filter(mdf.col("vote_count")>100).select(explode(from_json(mdf.col("production_companies"), jsonSchema)).as("production_companies"),$"vote_average").select($"production_companies.name",$"vote_average").rdd.map(v => (v(0).toString,(v(1).toString.toFloat,1))).repartition(1).reduceByKey((x,y) => (x._1+y._1 , x._2+y._2)).mapValues(x => (x._1/x._2,x._2)).collect().toList.map { case(company,(average,count)) =>(("company" ->company.replace("[","").replace("]",""))~("average" ->average)~("count" ->count))}val mdfJson=compact(render(jsonString))mdfJson }val str=moviesVote() println(str) val path="/home/chenbengang/ziyu_bigdata/quick_learn_spark/movie_moviesVote.txt" val hdfspath="hdfs://192.168.1.30:9000/user/root/movie_moviesVote.txt" save(str,path,hdfspath)

2.5. 電影預(yù)算和營收的關(guān)系

這部分考慮電影的營收情況，因此對于每個電影，需要導(dǎo)出如下的數(shù)據(jù)：

• [電影標(biāo)題，預(yù)算，收入]
  基于 DataFrame 對數(shù)據(jù)進(jìn)行字段過濾即可，過濾掉預(yù)算，收入為 0 的數(shù)據(jù)。：

// 5. 電影預(yù)算和營收的關(guān)系 // 3. 流行度和評價的關(guān)系 def budgetRevenue():Array[String]={mdf.select($"title",$"budget",$"revenue").filter(!$"budget".equalTo(0)).filter(!$"revenue".equalTo(0)).toJSON.collect() } val budgetRevenueArr=budgetRevenue() println(budgetVoteArr.length) val budgetRevenueSB = new StringBuilder budgetRevenueSB ++= "[" for (v <- budgetRevenueArr){budgetRevenueSB ++= v } budgetRevenueSB ++= "]" println(budgetRevenueSB.toString) val path="/home/chenbengang/ziyu_bigdata/quick_learn_spark/movie_budgetRevenue.txt" val hdfspath="hdfs://192.168.1.30:9000/user/root/movie_budgetRevenue.txt" save(budgetRevenueSB.toString,path,hdfspath)

五、數(shù)據(jù)可視化方法

數(shù)據(jù)可視化基于阿里開源的數(shù)據(jù)可視化工具 G2 實(shí)現(xiàn)。G2 是一套基于可視化編碼的圖形語法，以數(shù)據(jù)驅(qū)動，具有高度的易用性和擴(kuò)展性，用戶無需關(guān)注各種繁瑣的實(shí)現(xiàn)細(xì)節(jié)，一條語句即可構(gòu)建出各種各樣的可交互的統(tǒng)計圖表。下面以 TMDb 中電影的體裁分布為例說明可視化過程。

• 1.首先使用 python Web 框架 bottle 訪問可視化頁面方便進(jìn)行 json 數(shù)據(jù)的讀取。使用下面代碼web.py 可以實(shí)現(xiàn)一個簡單的靜態(tài)文件讀取。
• 2.bottle 對于接收到的請求進(jìn)行路由

對于 web 服務(wù)啟動目錄中 static 文件夾下的文件，直接返回對應(yīng)文件名的文件；

對于啟動目錄下的 html 文件，也返回對應(yīng)的頁面。

直接訪問本機(jī)的 9999 端口，則返回主頁。

• 3.最后，將 web 服務(wù)綁定到本機(jī)的 9999 端口。根據(jù)上面的實(shí)現(xiàn)，對于 web 頁面 (html 文件)，直接放在服務(wù)啟動的目錄下，對于 Spark 分析的結(jié)果，則保存在 static 目錄下。
  接下來實(shí)現(xiàn)主頁文件 index.html。
  （可視化代碼見pyspark代碼中的可視化部分）：百度網(wǎng)盤地址：https://pan.baidu.com/s/1lt7PHF17-gHieOU0B0zJ3A 提取碼：cui7

六、數(shù)據(jù)圖表

（一）概覽

1.TMDb 電影中的體裁分布：

從圖中可以看出，Drama 的電影在 TMDb 中占比較大，其次 Science Fiction、Action 和 Thriller 的數(shù)量也較多。

2.前 100 個常見關(guān)鍵詞

TMDb 中最常見的關(guān)鍵詞是 Woman Director，其次還有 independent film 等。

3. TMDb 中最常見的 10 種預(yù)算數(shù)

有 144 部電影的預(yù)算為 20,000,000，是最常見的預(yù)算值。

4. TMDb 中最常見電影時長 (只展示電影數(shù)大于 100 的時長)

多數(shù)電影的時長是90分鐘或100分鐘。

5. 生產(chǎn)電影最多的 10 大公司

生產(chǎn)電影較多的公司是 Warner Bros.、Universal Pictures等。

6. TMDb 中的 10 大電影語言

大多數(shù)電影中的語言是英語。

（二）關(guān)系

1.預(yù)算與評價的關(guān)系

預(yù)算高的電影不見得能取得更好的評價，例如預(yù)算高達(dá) 380,000,000 美元的 Pirates of the Caribbean: On Stranger Tides（加勒比海盜）評價只有6.4分。

2.發(fā)行時間與評價的關(guān)系

早期的電影評價都比較高，例如發(fā)行于1936年的 Modern Times（摩登時代）評價高達(dá)8.1分。

3.流行度和評價的關(guān)系

流行度較高的話一般能取得平均水平以上的評價，例如 Interstellar（星際穿越）流行度很高，評價為8.1分。

4.公司生產(chǎn)的電影平均分和數(shù)量的關(guān)系

從圖中可以看出，一個公司生產(chǎn)的電影越多，其電影平均分越接近整體的平均水平。

5.電影預(yù)算和營收的關(guān)系

從圖中可以看出，多數(shù)電影都能實(shí)現(xiàn)正收入，而預(yù)算為 237,000,000 美元的 Avatar（阿凡達(dá)）最終收入為2787,965,087美元

總結(jié)

以上是生活随笔為你收集整理的Spark大数据-TMDB电影数据分析（spark-scala版）的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。