java - 如何在文本文件中查找键并使用 Spark 和 JSON 将它们与另一个文件进行比较

Question

我目前正在学习 Spark，遇到一个问题，说给两个文本文件找到文本评论超过 100 个单词的书籍，然后过滤结果以仅显示恐怖类别。

这是我的两个文本文件的示例。
BookInformation.data:

在此数据文件中，我有 4 个 key 。

userName, price, categories, title

每个键都有一个值，并且每个键由 , 作为分隔符分隔。某些键使用字符串值，而其他键使用整数值。

{"username": "JAMES250", "price": 19.20, "categories": "Horror", "title": "Friday the 13th"}
{"username": "Bro2KXA1", "price": 09.21, "categories": "Fantasy", "title": "Wizard of Oz"}
{"username": "LucyLu1272", "price": 18.69, "categories": "Fiction", "title": "Not Real"}
{"username": "6302049040", "price": 08.86, "categories": "Fantasy", "title": "Fantastic"} 
...
etc
...

ReviewerText.data

在此数据文件中，我有 5 个 key 。

reviewerID, userName, reviewText, overall, reviewTime

每个键都有一个值，并且每个键由 , 作为分隔符分隔。某些键使用字符串值，而其他键使用整数值。

{"reviewerID": "A1R3P8MRFSN4X3", "username": "JAMES250", "reviewText": "Wow what a book blah blah… END", "overall": 4.0, "reviewTime": "08 9, 1997"}
{"reviewerID": "AVM91SKZ9M58T", " username ": " Bro2KXA1 ", "reviewText": "Different Blah Blah Blah Blah… So on… END", "overall": 5.0, "reviewTime": "08 10, 1997"}
{"reviewerID": "A1HC72VDRLANIW", " username ": "DiffUser09", "reviewText": "Another Review Blah Blah Blah Blah… So on… END", "overall": 1.0, "reviewTime": "08 19, 1997"}
{"reviewerID": "A2XBTS97FERY2Q", " username ": "MyNameIs01", "reviewText": "I love books. END", "overall": 5.0, "reviewTime": "08 23, 1997"}
...
etc
...

我的目标很简单。

1. 首先，我想检查 ReviewInformation.data 中是否存在超过 100 个单词的 reviewText。
2. 找到每一个超过 100 个单词的 reviewText 后，我想按照总体评分的顺序对结果进行排序；从 5 开始到 1。然后我还需要为每个打印相应的 Title 。
3. 之后，我需要重新启动过滤器，并且只需要从 BookInformation.data 中过滤掉类别，以仅显示恐怖类别。
4. 然后计算 reviewText 中出现的 恐怖 类别的平均单词数。

代码:到目前为止，我正在为每个文件中的每个行条目创建一个 Key:Value 数组。这里的目标是创建一个可以解析任何键并接收其值的数组。

package main.scala

import org.apache.spark.{SparkConf, SparkContext}
import scala.io.StdIn.readLine
import scala.io.Source

object ReviewDataSpark {
  def main(args: Array[String]) {
    //Create a SparkContext to initialize Spark
    val conf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Word Count")
    val sc = new SparkContext(conf)

    val metaDataFile = sc.textFile("/src/main/resources/BookInformation.data")
    val reviewDataFile = sc.textFile("/src/main/resources/ReviewText.data")

    reviewDataFile.flatMap { line => {
      val Array(label, rest) = line split ","
      println(Array)
      val items = rest.trim.split("\\s+")
      println(items)
      items.map(item => (label.trim -> item))
     } 
    }

    metaDataFile.flatMap { line => {
      val Array(label, rest) = line split ","
      println(Array)
      val items = rest.trim.split("\\s+")
      println(items)
      items.map(item => (label.trim -> item))
     }
    }

  }
}

问题:所以我的代码的主要问题是我不相信我正确使用了 flatMap。我似乎无法将键和值溢出到键数组中。

我的代码刚刚打印出来:进程已完成，退出代码为 0

这似乎不正确。

编辑:

所以我更新了我的代码以使用 JSON 库。

val jsonColName = "json"
// intermediate column name where we place each line of source data
val jsonCol = col(jsonColName) // its reusable ref
val metaDataSet = spark.read.textFile("src/main/resources/BookInformation.data")
  .toDF(jsonColName).select(get_json_object(jsonCol, "$.username")
  .alias("username"), get_json_object(jsonCol, "$.price")
  .alias("price"), get_json_object(jsonCol, "$.categories")
  .alias("categories"), get_json_object(jsonCol, "$.title")
  .alias("title"))

val reviewDataSet = spark.read.textFile("src/main/resources/reviewText.data")
  .toDF(jsonColName).select(get_json_object(jsonCol, "$.reviewerID")
  .alias("reviewerID"), get_json_object(jsonCol, "$.username")
  .alias("username"), get_json_object(jsonCol, "$.reviewText")
  .alias("reviewText"), get_json_object(jsonCol, "$.overall")
  .alias("overall").as[Double], get_json_object(jsonCol, "$.reviewTime")
  .alias("reviewTime"))
reviewDataSet.show()
metaDataSet.show()

多亏了这些信息，我才能够合并。

val joinedDataSets = metaDataSet.join(reviewDataSet, Seq("username"))
    joinedDataSets.show()

现在，我的下一步是能够计算 ReviewText 列中 joinedDataSets 内的单词数，并仅保留超过 100 个单词的单词。

如何从键 reviewText 中过滤 JSON 对象，然后计算所有条目并删除少于 100 个单词的条目。

Answer 1

首先，您需要以结构化方式从文件中加载数据。源文件的每一行都可以解析为 JSON，并且信息应正确放置在相应的列中。

例如，加载和解析BookInformation.data:

import org.apache.spark.sql.functions._ // necessary for col, get_json_object functions and others below

val session = SparkSession.builder().appName("My app")
                         .master("local[*]")
                         .getOrCreate()

val bookInfoFilePath = // path to BookInformation.data

val jsonColName = "json" // intermediate column name where we place each line of source data
val jsonCol = col(jsonColName) // its reusable ref

val bookInfoDf = session.read.textFile(bookInfoFilePath).toDF(jsonColName).select(
  get_json_object(jsonCol, "$.username").alias("username"),
  get_json_object(jsonCol, "$.price").alias("price"),
  get_json_object(jsonCol, "$.categories").alias("categories"),
  get_json_object(jsonCol, "$.title").alias("title")
)

现在我们有一个包含正确结构化数据的图书信息 DataFrame:

bookInfoDf.show()
+----------+-----+----------+---------------+
|  username|price|categories|          title|
+----------+-----+----------+---------------+
|  JAMES250| 19.2|    Horror|Friday the 13th|
|  Bro2KXA1| 9.21|   Fantasy|   Wizard of Oz|
|LucyLu1272|18.69|   Fiction|       Not Real|
|6302049040| 8.86|   Fantasy|      Fantastic|
+----------+-----+----------+---------------+

Q3 和 Q4 的答案变得显而易见。

val dfQuestion3 = bookInfoDf.where($"categories" === "Horror")
dfQuestion3.show()
+--------+-----+----------+---------------+
|username|price|categories|          title|
+--------+-----+----------+---------------+
|JAMES250| 19.2|    Horror|Friday the 13th|
+--------+-----+----------+---------------+

对于第四季度，您必须使用 username 列将 bookInfoDf 与从 ReviewerText.data 加载的 DataFrame 连接起来，然后聚合( .agg)reviewText 列的平均长度数据(avg 和 length 函数)。

要加载ReviewerText.data，您可以按照上面加载bookInfoDf的方式进行类比。 overall 列应在 .alias 调用后使用 .as[Double] 转换为数字。

更新

I had a question about how to count the number of words within a JSON Key/Value. For example, in the key reviewText I have create and merged both BookInformation and ReviewText into one dataset. Now If i wanted to loop through each reviewText and count the number of words then filter either keep or remove depending on the amount of words within the Key's Value how would I go about doing that? I'm trying to learn how to extract value

一种可能的方法是计算单词数并将其存储在专用列中:

// reviewerTextDf is the DataFrame with original data from ReviewerText.data
val dfWithReviewWordsCount = reviewerTextDf.withColumn("nb_words_review", size(split($"reviewText", "\\s+")))

dfWithReviewWordsCount.show()

给出以下内容:

+--------------+--------+--------------------+-------+-----------+---------------+
|    reviewerID|username|          reviewText|overall| reviewTime|nb_words_review|
+--------------+--------+--------------------+-------+-----------+---------------+
|A1R3P8MRFSN4X3|JAMES250|Wow what a book b...|    4.0| 08 9, 1997|              7|
| AVM91SKZ9M58T|    null|Different Blah Bl...|    5.0|08 10, 1997|              8|
|A1HC72VDRLANIW|    null|Another Review Bl...|    1.0|08 19, 1997|              9|
|A2XBTS97FERY2Q|    null|   I love books. END|    5.0|08 23, 1997|              4|
+--------------+--------+--------------------+-------+-----------+---------------+