TMDB電影數據分析（spark-scala版，pyspark版本）

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

一.環境要求（僅供參考）

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

二.數據預處理

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

• 對數據去除標題行。數據中某些字段包含 json 數據，直接使用 DataFrame 進行讀取會出現分割錯誤，如果要創建 DataFrame，先讀取文件生成 RDD，再將 RDD 轉為 DataFrame。
• 為了便于處理上傳數據至hdfs文件系統：

hdfs dfs -put tmdb_5000_movies.csv

三、使用 Spark 將數據轉為 DataFrame

為了創建 DataFrame，首先需要將 HDFS 上的數據加載成 RDD，再將 RDD 轉化為 DataFrame。下面代碼段完成從文件到 RDD 再到 DataFrame 的轉化：

// 創建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" // 根據字符串生成模式 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數據某些字段中包含 val mdf=spark.read.format("com.databricks.spark.csv").schema(schema).option("inferSchema", value = false).option("header", value = true).option("nullValue", "\\N").option("escape", "\"") // 設置用于在已引用的值內轉義引號的單個字符。詳情見 spark 讀取 csv 官網介紹 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 進行數據分析

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

1.概覽

這個部分對數據進行整體的分析。

1.1.TMDb 電影中的體裁分布

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

// （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解釋器執行上面語句后返回的結果 // // 尋找tmdb中最常見的10中預算數，對電影預算頻率進行統計 // // 使用sql語句分析 // 或者使用spark的轉換函數進行分析 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) // 此處是單條中包含多個數據的json，按照jsonSchema的格式進行解析，并生成多條單個數據，explode是將數組組生成為列。.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) }

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

基于這個函數實現 countByGenres 用來生成不同體裁的電影數統計結果，接著，使用下面代碼進行數據導出至 genres.json 方便之后進行可視化：

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文件系統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 個常見關鍵詞

該項分析電影關鍵詞中出現頻率最高的前一百個。由于關鍵詞字段也是 json 格式數據，因此調用 countByJson 進行頻率統計，同時對于統計結果進行降序排序并取前 100 項即可：

// 2. 前 100 個常見關鍵詞 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 種預算數

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

// 3. TMDb 中最常見的 10 種預算數 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 中最常見電影時長 (只展示電影數大于 100 的時長)

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

// 4. TMDb 中最常見電影時長 (只展示電影數大于 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. 生產電影最多的 10 大公司

這一項統計電影產出最多的 10 個公司，同樣使用 countByJson 對 JSON 數據進行頻率統計，然后進行降序排列取前 10 項即可。

// 5. 生產電影最多的 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 大電影語言

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

// 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.關系

這個部分考慮數據之間的關系。

2.1.預算與評價的關系

這部分考慮預算與評價之間的關系，因此對于每個電影，需要導出如下的數據：

• [電影標題，預算，評價]

// 2.關系 // 這個部分考慮數據之間的關系。 // 1. 預算與評價的關系 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. 發行時間與評價的關系

這部分考慮發行時間與評價之間的關系，因此對于每個電影，需要導出如下的數據：

• [電影標題，發行時間，評價]

// 2. 發行時間與評價的關系 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. 流行度和評價的關系

這部分考慮流行度與評價之間的關系，因此對于每個電影，需要導出如下的數據：

• [電影標題，流行度，評價]

// 3. 流行度和評價的關系 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. 公司生產的電影平均分和數量的關系

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

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

// 4. 公司生產的電影平均分和數量的關系 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. 電影預算和營收的關系

這部分考慮電影的營收情況，因此對于每個電影，需要導出如下的數據：

• [電影標題，預算，收入]
  基于 DataFrame 對數據進行字段過濾即可，過濾掉預算，收入為 0 的數據。：

// 5. 電影預算和營收的關系 // 3. 流行度和評價的關系 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)

五、數據可視化方法

數據可視化基于阿里開源的數據可視化工具 G2 實現。G2 是一套基于可視化編碼的圖形語法，以數據驅動，具有高度的易用性和擴展性，用戶無需關注各種繁瑣的實現細節，一條語句即可構建出各種各樣的可交互的統計圖表。下面以 TMDb 中電影的體裁分布為例說明可視化過程。

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

對于 web 服務啟動目錄中 static 文件夾下的文件，直接返回對應文件名的文件；

對于啟動目錄下的 html 文件，也返回對應的頁面。

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

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

六、數據圖表

（一）概覽

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

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

2.前 100 個常見關鍵詞

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

3. TMDb 中最常見的 10 種預算數

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

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

多數電影的時長是90分鐘或100分鐘。

5. 生產電影最多的 10 大公司

生產電影較多的公司是 Warner Bros.、Universal Pictures等。

6. TMDb 中的 10 大電影語言

大多數電影中的語言是英語。

（二）關系

1.預算與評價的關系

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

2.發行時間與評價的關系

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

3.流行度和評價的關系

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

4.公司生產的電影平均分和數量的關系

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

5.電影預算和營收的關系

從圖中可以看出，多數電影都能實現正收入，而預算為 237,000,000 美元的 Avatar（阿凡達）最終收入為2787,965,087美元

總結

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

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。