我试着用spark加载一个包含电影元数据的tsv文件，这个tsv文件包含json格式的电影类型信息[每行的最后一列]
示例文件
975900 /m/03vyhn Ghosts of Mars 2001-08-24 14010832 98.0 {"/m/02h40lc": "English Language"} {"/m/09c7w0": "United States of America"} {"/m/01jfsb": "Thriller", "/m/06n90": "Science Fiction", "/m/03npn": "Horror", "/m/03k9fj": "Adventure", "/m/0fdjb": "Supernatural", "/m/02kdv5l": "Action", "/m/09zvmj": "Space western"}
3196793 /m/08yl5d Getting Away with Murder: The JonBenét Ramsey Mystery 2000-02-16 95.0 {"/m/02h40lc": "English Language"} {"/m/09c7w0": "United States of America"} {"/m/02n4kr": "Mystery", "/m/03bxz7": "Biographical film", "/m/07s9rl0": "Drama", "/m/0hj3n01": "Crime Drama"}我尝试了下面的代码，它使我能够从类型json访问特定的值
val ss = SessionCreator.createSession("DataCleaning", "local[*]")//helper function creates a spark session and returns it
val headerInfoRb = ResourceBundle.getBundle("conf.headerInfo")
val movieDF = DataReader.readFromTsv(ss, "D:/Utility/Datasets/MovieSummaries/movie.metadata.tsv")
.toDF(headerInfoRb.getString("metadataReader").split(',').toSeq:_*)//Datareader.readFromTsv is a helper function to read TSV file ,takes sparkSession and file path as input to resurn a dataframe, which uses sparkSession's read function
movieDF.select("wiki_mv_id","mv_nm","mv_genre")
.withColumn("genre_frmttd", get_json_object(col("mv_genre"), "$./m/02kdv5l"))
.show(1,false)输出
+----------+--------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------+
|wiki_mv_id|mv_nm |mv_genre |genre_frmttd|
+----------+--------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------+
|975900 |Ghosts of Mars|{"/m/01jfsb": "Thriller", "/m/06n90": "Science Fiction", "/m/03npn": "Horror", "/m/03k9fj": "Adventure", "/m/0fdjb": "Supernatural", "/m/02kdv5l": "Action", "/m/09zvmj": "Space western"}|Action |
+----------+--------------+------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+------------+
only showing top 1 row我希望数据框中的每一行都有一个frmtd列，如下所示[下面的代码片段是第一个示例行]
[Thriller,Fiction,Horror,Adventure,Supernatural,Action,Space Western]我是scala和spark的一个新手，确实建议一些方法来列出这些值
1条答案
按热度按时间bn31dyow1#
使用解析json
from_json投给MapType(StringType, StringType)仅提取值使用map_values```import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.{MapType, StringType}
movieDF.select("wiki_mv_id","mv_nm","mv_genre")
.withColumn("genre_frmttd",map_values(from_json(col("mv_genre"),MapType(StringType, StringType))))
.show(1,false)