毕设三: spark与phoenix集成插入数据/解析json数组

需求:将前些日子采集的评论存储到hbase中

思路:

先用fastjson解析评论,然后构造rdd,最后使用spark与phoenix交互,把数据存储到hbase中

部分数据:

[
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G全网通手机 双卡双待",
    "creationTime": "2019-04-08 01:13:42",
    "content": "此用户没有填写评价内容",
    "label": []
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G全网通手机 双卡双待",
    "creationTime": "2019-03-29 11:49:36",
    "content": "不错",
    "label": []
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G全网通手机 双卡双待",
    "creationTime": "2019-03-21 21:13:07",
    "content": "正品没什么毛病。信号好像是照别的差一点，但是还可以，不是特别差。分辨率不是那么好，但是也不是特别差。一般般。手机不卡。打游戏很顺畅。官方正品没有翻车。",
    "label": [
      {
        "labelName": "功能齐全"
      }
    ]
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G全网通手机 双卡双待",
    "creationTime": "2019-03-22 09:56:22",
    "content": "不错是正品",
    "label": [
      {
        "labelName": "系统流畅"
      },
      {
        "labelName": "声音大"
      },
      {
        "labelName": "做工精致"
      },
      {
        "labelName": "待机时间长"
      }
    ]
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G全网通手机 双卡双待",
    "creationTime": "2019-03-13 07:27:56",
    "content": "性价比很高的手机，用习惯了ios转安卓7个月，重新回归，只能说，用苹果省心。苏宁质量有保障，送货快，价格优惠，推荐购买！",
    "label": [
      {
        "labelName": "系统流畅"
      },
      {
        "labelName": "功能齐全"
      },
      {
        "labelName": "包装一般"
      }
    ]
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G手机 双卡双待",
    "creationTime": "2019-02-25 22:03:18",
    "content": "弟弟就想要一个苹果手机。本来打算买8p的。然后我，推荐他这款xr，是最新款。价格总体来说性价比，比8p好。买了很快就到了，第二天就。屏幕很大，还是面容id。苹果x大很多。比***x小一点",
    "label": [
      {
        "labelName": "外观漂亮"
      },
      {
        "labelName": "做工精致"
      },
      {
        "labelName": "反应快"
      },
      {
        "labelName": "系统流畅"
      }
    ]
  },
  {
    "referenceName": "【券后低至5388】Apple iPhone XR 64GB 黑色 移动联通电信4G手机 双卡双待",
    "creationTime": "2019-02-21 12:45:22",
    "content": "物流很棒，xr价格可以接受、性能稳定！",
    "label": []
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G全网通手机 双卡双待",
    "creationTime": "2019-02-22 12:14:55",
    "content": "很不错的手机除了有点厚，极致的体验，黑边完全没有影响",
    "label": [
      {
        "labelName": "拍照效果好"
      },
      {
        "labelName": "待机时间长"
      },
      {
        "labelName": "电池耐用"
      },
      {
        "labelName": "性价比高 "
      }
    ]
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G手机 双卡双待",
    "creationTime": "2019-02-13 00:28:03",
    "content": "非常非常好的商品。很不错，下次再来",
    "label": [
      {
        "labelName": "信号稳定"
      },
      {
        "labelName": "反应快"
      },
      {
        "labelName": "声音大"
      },
      {
        "labelName": "做工精致"
      }
    ]
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G 手机",
    "creationTime": "2019-04-02 17:29:43",
    "content": "此用户没有填写评价内容",
    "label": []
  }
]
[
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G全网通手机 双卡双待",
    "creationTime": "2019-04-05 18:13:14",
    "content": "满意嘻嘻",
    "label": [
      {
        "labelName": "音质好"
      },
      {
        "labelName": "拍照效果好"
      },
      {
        "labelName": "功能齐全"
      },
      {
        "labelName": "外观漂亮"
      }
    ]
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G全网通手机 双卡双待",
    "creationTime": "2019-03-21 00:13:17",
    "content": "棒棒哒",
    "label": []
  },
  {
    "referenceName": "【双12爆款】Apple iPhone XR 64GB 黑色 移动联通电信4G手机 双卡双待",
    "creationTime": "2019-01-19 10:23:57",
    "content": "双十二买的正好赶上手机坏了就买了xr太贵没舍得买一直用苹果的系统用顺手了不愿意换",
    "label": [
      {
        "labelName": "反应快"
      },
      {
        "labelName": "做工精致"
      },
      {
        "labelName": "信号稳定"
      },
      {
        "labelName": "待机时间长"
      },
      {
        "labelName": "性价比一般般"
      }
    ]
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G全网通手机 双卡双待",
    "creationTime": "2019-03-21 12:47:34",
    "content": "用了几天感觉还可以，只是信号不是很好，总体上是可以的",
    "label": [
      {
        "labelName": "音质好"
      },
      {
        "labelName": "分辨率高"
      }
    ]
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G全网通手机 双卡双待",
    "creationTime": "2019-03-09 08:32:15",
    "content": "苹果手机做工精细，手感不错，外观设计也是很有时尚感！系统运行十分流畅，操作体验不错；功能齐全，屏幕分辨率高，总体来说很满意！",
    "label": [
      {
        "labelName": "做工精致"
      },
      {
        "labelName": "系统流畅"
      },
      {
        "labelName": "功能齐全"
      }
    ]
  },
  {
    "referenceName": "【低至5399】Apple iPhone XR 64GB 黑色 移动联通电信4G手机 双卡双待",
    "creationTime": "2019-01-15 22:44:53",
    "content": "真心喜欢，一直在徘徊x.还是xr.个人觉得真不错，一直信赖苏宁，新机为激活，价钱能接受，值得推荐，黑边什么的不影响。",
    "label": [
      {
        "labelName": "拍照效果好"
      },
      {
        "labelName": "外观漂亮"
      },
      {
        "labelName": "屏幕清晰"
      }
    ]
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G全网通手机 双卡双待",
    "creationTime": "2019-03-08 22:12:18",
    "content": "手机运行流畅，外观也漂亮，黑色彰显档次，音质也好，又是双卡双待手机，性价比不错，比起XS便宜不少。",
    "label": [
      {
        "labelName": "外观漂亮"
      },
      {
        "labelName": "系统流畅"
      }
    ]
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G手机 双卡双待",
    "creationTime": "2019-03-01 15:54:33",
    "content": "手机很好，电池非常抗用，价钱也非常美丽，值得购买。",
    "label": [
      {
        "labelName": "系统流畅"
      },
      {
        "labelName": "电池耐用"
      },
      {
        "labelName": "分辨率高"
      },
      {
        "labelName": "待机时间长"
      },
      {
        "labelName": "包装一般"
      }
    ]
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G手机 双卡双待",
    "creationTime": "2019-02-19 09:37:25",
    "content": "春节期间配送超快，手机没有任何问题，苏宁易购确实做到了全网最低价",
    "label": []
  },
  {
    "referenceName": "Apple iPhone XR 64GB 黑色 移动联通电信4G全网通手机 双卡双待",
    "creationTime": "2019-02-23 13:27:15",
    "content": "挺好的懒得拍照了借了几张黑边确实大",
    "label": [
      {
        "labelName": "待机时间长"
      },
      {
        "labelName": "反应快"
      },
      {
        "labelName": "做工精致"
      }
    ]
  }
]

数据是json数组,因此采用fastjson进行解析,为了最终存数据的更方便,需要构造Comment类

依赖

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.47</version>
        </dependency>
        
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.9</version>
        </dependency>
    
    
        <!-- hbase -->
        <!-- <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>2.0.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-server</artifactId>
            <version>2.0.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-common</artifactId>
            <version>2.0.2</version>
        </dependency> -->
        
        
        <!-- phoenix -->
        <dependency>
            <groupId>org.apache.phoenix</groupId>
            <artifactId>phoenix-core</artifactId>
            <version>5.0.0-HBase-2.0</version>
        </dependency>
        
         
        <!-- phoenix_spark -->
         <dependency>
            <groupId>org.apache.phoenix</groupId>
            <artifactId>phoenix-spark</artifactId>
            <version>5.0.0-HBase-2.0</version>
        </dependency>
         
         
         <!-- spark -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.0.2</version>
        </dependency>
          
          
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.11</artifactId>
            <version>2.0.2</version>
        </dependency>

 Comment.java

package cn.tele.bean;

/**
 *
 * @author Tele
 *
 */
public class Comment {
    private Integer id;
    private String name;
    private String content;
    private String creationtime;
    private String label;
    private String platform;

    public Integer getId() {
        return id;
    }

    public void setId(Integer id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }

    public String getCreationtime() {
        return creationtime;
    }

    public void setCreationtime(String creationtime) {
        this.creationtime = creationtime;
    }

    public String getLabel() {
        return label;
    }

    public void setLabel(String label) {
        this.label = label;
    }

    public String getPlatform() {
        return platform;
    }

    public void setPlatform(String platform) {
        this.platform = platform;
    }
}

StoreData.java

package cn.tele.spark;

import java.io.IOException;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.UUID;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.storage.StorageLevel;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import cn.tele.bean.Comment;
import scala.Tuple2;

/**
 * 存储爬取的评论到hbase中
 * 
 * @author Tele
 *
 */
public class StoreData {
    private static SparkConf conf = new SparkConf().setAppName("storedata").setMaster("local").set("spark.serializer",
            "org.apache.spark.serializer.KryoSerializer");
    private static JavaSparkContext jsc = new JavaSparkContext(conf);
    private static SparkSession session = new SparkSession(jsc.sc());
    static {
        // 注册
        conf.registerKryoClasses(new Class[] { Comment.class });
    }
    // 链接信息
    private static final String DB_PHOENIX_URL = "jdbc:phoenix:hadoop002,hadoop003,hadoop004";
    /*
     * private static final String DB_PHOENIX_DRIVER =
     * "org.apache.phoenix.jdbc.PhoenixDriver"; private static final String
     * DB_PHOENIX_USER = ""; private static final String DB_PHOENIX_PASS = "";
     * private static final String DB_PHOENIX_FETCHSIZE = "10000";
     */

    public static void main(String[] args) throws SQLException {

        // 遍历文件夹
        Path path = Paths.get("F:\comment\");

        try {
            MyFileVisitor myFileVisitor = new MyFileVisitor();
            Files.walkFileTree(path, myFileVisitor);
            List<Map<String, Object>> list = myFileVisitor.getData();
            JavaRDD<Comment> commentRDD = getCommentRDD(list);
            // 存储至hbase
            storeData(commentRDD);

        } catch (IOException e) {
            e.printStackTrace();
        }

        // 读取数据
        /*
         * JavaRDD<String> rdd =
         * jsc.textFile("file:\F:\comment\sn_comment\iphonexr-2019-04-16-18-27-36\"
         * ); List<Comment> commentList = getCommentList(rdd,"iphonexr");
         */

        jsc.close();

    }

    private static int storeData(JavaRDD<Comment> rdd) {
        int successCount = 0;
        /*
         * DataTypes.createStructType(Arrays.asList(
         * DataTypes.createStructField("id",DataTypes.IntegerType,false),
         * DataTypes.createStructField("name",DataTypes.StringType,false),
         * DataTypes.createStructField("content",DataTypes.StringType,false),
         * DataTypes.createStructField("creationtime",DataTypes.StringType,false),
         * DataTypes.createStructField("label",DataTypes.StringType,true),
         * DataTypes.createStructField("platform",DataTypes.StringType,false) ));
         */

        Dataset<Row> ds = session.createDataFrame(rdd, Comment.class);
        ds.write().mode(SaveMode.Overwrite).format("org.apache.phoenix.spark").option("zkUrl", DB_PHOENIX_URL)
                .option("table", "comment").save();
        ;

        return successCount;
    }

    @SuppressWarnings("unchecked")
    private static <U> JavaRDD<Comment> getCommentRDD(List<Map<String, Object>> list) {
        JavaRDD<Map<String, Object>> originalRDD = jsc.parallelize(list);
        JavaRDD<List<Comment>> listCommentRDD = originalRDD.map(new Function<Map<String, Object>, List<Comment>>() {

            private static final long serialVersionUID = 1L;

            List<Comment> dataList = new ArrayList<>();

            public List<Comment> call(Map<String, Object> v1) throws Exception {
                Set<Entry<String, Object>> entrySet = v1.entrySet();
                Iterator<Entry<String, Object>> iterator = entrySet.iterator();
                while (iterator.hasNext()) {
                    Entry<String, Object> entry = iterator.next();
                    String referenceName = entry.getKey();
                    String platform = referenceName.split("#")[0];
                    List<Comment> commentList = (List<Comment>) entry.getValue();
                    commentList.forEach(cm -> {
                        cm.setPlatform(platform);
                        dataList.add(cm);
                    });
                    println(referenceName + "评论量------------------:" + commentList.size());
                }
                return dataList;
            }
        }).persist(StorageLevel.MEMORY_ONLY());

        JavaRDD<Comment> commentRDD = listCommentRDD.flatMap(new FlatMapFunction<List<Comment>, Comment>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Iterator<Comment> call(List<Comment> t) throws Exception {
                return t.iterator();
            }
        });

        long totalSize = commentRDD.count();
        println("评论总量:-----------" + totalSize);

        // 设置id
        JavaRDD<Comment> resultRDD = commentRDD.zipWithIndex().map(new Function<Tuple2<Comment, Long>, Comment>() {

            private static final long serialVersionUID = 1L;

            @Override
            public Comment call(Tuple2<Comment, Long> v1) throws Exception {
                v1._1.setId(Integer.valueOf(v1._2.toString()));
                return v1._1;
            }
        });
        return resultRDD;
    }

    private static List<Comment> getCommentList(JavaRDD<String> rdd, String referenceName) {
        List<Comment> commentList = new ArrayList<Comment>();

        String originalStr = rdd.reduce(new Function2<String, String, String>() {
            private static final long serialVersionUID = 1L;

            public String call(String v1, String v2) throws Exception {
                return v1.trim() + v2.trim();
            }
        });
        String uuid = UUID.randomUUID().toString();
        originalStr = originalStr.replace("][", "]" + uuid + "[");

        // 解析json
        String[] pages = originalStr.split(uuid);
        for (String page : pages) {
            try {
                JSONArray jsonArray = JSON.parseArray(page);
                for (Object obj : jsonArray) {
                    JSONObject jsonObject = (JSONObject) obj;
//                    String referenceName  = jsonObject.getString("referenceName");
                    String creationTime = jsonObject.getString("creationTime");
                    String content = jsonObject.getString("content");
                    println("referenceName:" + referenceName);
                    println("creationTime:" + creationTime);
                    println("content:" + content);

                    // 封装
                    Comment comment = new Comment();
                    comment.setName(referenceName);
                    comment.setCreationtime(creationTime);
                    comment.setContent(content);

                    JSONArray labelArray = jsonObject.getJSONArray("label");
                    if (labelArray != null) {
                        String label = "";
                        for (Object labelObj : labelArray) {
                            JSONObject labelObject = (JSONObject) labelObj;
                            label += labelObject.getString("labelName") + "#";
                        }
                        comment.setLabel(label);
                        println("label:" + label);
                    }
                    commentList.add(comment);
                }
            } catch (Exception e) {
                continue;
            }
        }
        return commentList;
    }

    private static class MyFileVisitor extends SimpleFileVisitor<Path> {
        private List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();
        String platform = null;

        @Override
        public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
            println("当前访问的文件夹是------" + dir.toAbsolutePath() /* + dir.getFileName() */);
            String fileName = dir.getFileName().toString();
            if (fileName.contains("_")) {
                platform = fileName.split("_")[0];
            }
            if (platform != null) {
                if (fileName.contains("-")) {
                    String referenceName = fileName.split("-")[0];
                    JavaRDD<String> rdd = jsc.textFile(dir.toAbsolutePath().toString());
                    List<Comment> commentList = getCommentList(rdd, referenceName);
                    Map<String, Object> map = new HashMap<String, Object>();
                    // 平台_品牌--评论
                    map.put(platform + "#" + referenceName, commentList);
                    list.add(map);
                }
            }

            return super.preVisitDirectory(dir, attrs);
        }

        @Override
        public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
            return super.visitFile(file, attrs);
        }

        @Override
        public FileVisitResult visitFileFailed(Path file, IOException exc) throws IOException {
            return super.visitFileFailed(file, exc);
        }

        @Override
        public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
            return super.postVisitDirectory(dir, exc);
        }

        public List<Map<String, Object>> getData() {
            return list;
        }

    }

    private static void println(Object obj) {
        System.out.println(obj);
    }

}

可能会报如下异常,但并不影响,似乎和我使用的hbase版本有关

在终端进行查看

说明:

1.StorgeData还可以再进行优化,比如解析json的时候可以直接构造rdd而不是用list,也可以改造成集群上运行的版本,但由于我的数据量不多,直接用本地模式就足够了

2.SaveMode只能是SaveMode.Overwrite其他模式phoenix都不支持,实际测试时发现还是append

3.生成id时用了zipWithIndex,但连续的id容易造成集群热点问题,使用phoenix建表时最好加盐

4.与phoenix交互时也可以用spark的jdbc,可以参考https://blog.csdn.net/xiongbingcool/article/details/81458602

5.关于目录问题,spark支持对目录下的多个文件进行读取构造rdd,因此遍历时到父文件夹即可,当然,遍历到每个文件也可以

附测试用例:

/** 
 *
 *@author Tele
 *测试spark与phoenix集成
 */
public class SparkPhoneix {
    private static SparkConf conf = new SparkConf().setAppName("demo").setMaster("local");
    private static JavaSparkContext jsc = new JavaSparkContext(conf);
    private static SparkSession session = new SparkSession(jsc.sc());
    
    //链接信息
    private static final String DB_PHOENIX_URL = "jdbc:phoenix:hadoop002,hadoop003,hadoop004";
    /*private static final String DB_PHOENIX_DRIVER = "org.apache.phoenix.jdbc.PhoenixDriver";
    private static final String DB_PHOENIX_USER = "";
    private static final String DB_PHOENIX_PASS = "";
    private static final String DB_PHOENIX_FETCHSIZE = "10000";*/
    
    public static void main(String[] args) {
         Dataset<Row> ds = session.read().format("org.apache.phoenix.spark")
                                        .option("zkUrl", DB_PHOENIX_URL)
                                        .option("table", "TEST")
                                        .load();
         ds.createOrReplaceTempView("test");
         
         ds.show();
         
         
         //插入数据测试_SALT,ID,INFO.ITEM,INFO.CONTENT,INFO.LABEL
        StructType schema = DataTypes.createStructType(Arrays.asList(
                    DataTypes.createStructField("id",DataTypes.IntegerType,false),
                    DataTypes.createStructField("item",DataTypes.StringType,false),
                    DataTypes.createStructField("content",DataTypes.StringType,false),
                    DataTypes.createStructField("label",DataTypes.StringType,true)
                ));
        
        
        // 创建数据
        List<Row> list = new ArrayList<Row>();
        Row row1 = RowFactory.create(3,"iphone","不错",null);
        list.add(row1);
        
        Dataset<Row> dataset = session.createDataFrame(list,schema);
        //对于phoenix只能使用overwrite模式,但实际操作时发现是append数据
        dataset.write().mode(SaveMode.Overwrite).format("org.apache.phoenix.spark")
                                         .option("zkUrl", DB_PHOENIX_URL)
                                         .option("table", "TEST").save();;
        ds.show();
        session.stop();
        jsc.close();
    }
}