0. 说明
在 IDEA 中编写 Spark 代码实现将 JSON 数据转换成标签，分别用 Scala & Java 两种代码实现。
1. 准备
1.1 pom.xml
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.11</artifactId>
<version>2.1.0</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.47</version>
</dependency>
</dependencies>
1.2 工具类 TagUtil
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject; import java.util.ArrayList;
import java.util.List; /**
* 从 json 中抽取评论集合
*/
public class TagUtil {
public static List<String> extractTag(String json) { List<String> list = new ArrayList<String>(); // 将字符串解析成 json 对象
JSONObject obj = JSON.parseObject(json);
JSONArray arr = obj.getJSONArray("extInfoList");
if (arr != null && arr.size() > 0) {
// 得到数组的第一个 json 对象
JSONObject firstObj = arr.getJSONObject(0);
JSONArray values = firstObj.getJSONArray("values");
if (values != null && values.size() > 0) {
for (int i = 0; i < values.size(); i++) {
String tag = values.getString(i);
list.add(tag);
}
}
}
return list;
}
}
2. 标签生成代码编写
2.1 Scala 版
import java.util
import com.share.util.TagUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext} /**
* 标签生成
*/
object TaggenScala1 {
def main(args: Array[String]): Unit = {
// 创建 spark 配置对象
val conf = new SparkConf()
conf.setAppName("TaggenApp")
conf.setMaster("local") // 创建上下文
val sc = new SparkContext(conf) // 1. 加载文件
val rdd1 = sc.textFile("file:///e:/temptags.txt") // 2. 解析每行的json数据成为集合
val rdd2: RDD[(String, java.util.List[String])] = rdd1.map(line => {
val arr: Array[String] = line.split("\t")
// 商家id
val busid: String = arr(0)
// json
val json: String = arr(1)
val list: java.util.List[String] = TagUtil.extractTag(json)
Tuple2[String, java.util.List[String]](busid, list)
}) // 3. 过滤空集合 (85766086,[干净卫生, 服务热情, 价格实惠, 味道赞])
val rdd3: RDD[(String, util.List[String])] = rdd2.filter((t: Tuple2[String, java.util.List[String]]) => {
!t._2.isEmpty
}) // 4. 将值压扁 (78477325,味道赞)
val rdd4: RDD[(String, String)] = rdd3.flatMapValues((list: java.util.List[String]) => {
// 导入隐式转换
import scala.collection.JavaConversions._
list
}) // 5. 滤除数字的tag (78477325,菜品不错)
val rdd5 = rdd4.filter((t: Tuple2[String, String]) => {
try {
Integer.parseInt(t._2)
false
} catch {
case _ => true
}
}) // 6. 标1成对 ((70611801,环境优雅),1)
val rdd6: RDD[Tuple2[Tuple2[String, String], Int]] = rdd5.map((t: Tuple2[String, String]) => {
Tuple2[Tuple2[String, String], Int](t, 1)
}) // 7. 聚合 ((78477325,味道赞),8)
val rdd7: RDD[Tuple2[Tuple2[String, String], Int]] = rdd6.reduceByKey((a: Int, b: Int) => {
a + b
}) // 8. 重组 (83073343,List((性价比高,8)))
val rdd8: RDD[Tuple2[String, List[Tuple2[String, Int]]]] = rdd7.map((t: Tuple2[Tuple2[String, String], Int]) => {
Tuple2[String, List[Tuple2[String, Int]]](t._1._1, Tuple2[String, Int](t._1._2, t._2) :: Nil)
}) // 9. reduceByKey (71039150,List((环境优雅,1), (价格实惠,1), (朋友聚会,1), (团建,1), (体验好,1)))
val rdd9: RDD[Tuple2[String, List[Tuple2[String, Int]]]] = rdd8.reduceByKey((a: List[Tuple2[String, Int]], b: List[Tuple2[String, Int]]) => {
a ::: b
}) // 10. 分组内排序 (88496862,List((回头客,5), (服务热情,4), (味道赞,4), (分量足,3), (性价比高,2)))
val rdd10: RDD[Tuple2[String, List[Tuple2[String, Int]]]] = rdd9.mapValues((list: List[Tuple2[String, Int]]) => {
val list2: List[Tuple2[String, Int]] = list.sortBy((t: Tuple2[String, Int]) => {
-t._2
})
list2.take(5)
}) // 11. 商家间排序 (75144086,List((服务热情,38), (效果赞,30), (无办卡,22), (环境优雅,22), (性价比高,21)))
val rdd11: RDD[Tuple2[String, List[Tuple2[String, Int]]]] = rdd10.sortBy((t: Tuple2[String, List[Tuple2[String, Int]]]) => {
t._2(0)._2
}, false) rdd11.collect().foreach(println)
}
}
2.2 Java 版
待补充。。。