【待补充】[Spark Core] Spark 实现标签生成

---

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("	")
      // 商家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 版

待补充。。。

---