Flatmap三大功能

1，压平切分：

2，转换数据结构：

3，过滤：因为flatmap的返回值是void，所以数据可以直接写出去

  //TODO 3.将数据转换为JSON对象
        SingleOutputStreamOperator<JSONObject> jsonObjDS = pageStringDS.map(JSON::parseObject);

        //TODO 4.过滤数据,只需要访问主页跟商品详情页的数据
        SingleOutputStreamOperator<JSONObject> homeAndDetailPageDS = jsonObjDS.filter(new FilterFunction<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) throws Exception {
                String pageId = value.getJSONObject("page").getString("page_id");
                return "good_detail".equals(pageId) || "home".equals(pageId);
            }
        });

        //使用FlatMap代替
//        pageStringDS.flatMap(new FlatMapFunction<String, JSONObject>() {
//            @Override
//            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
//                JSONObject jsonObject = JSON.parseObject(value);
//                String pageId = jsonObject.getJSONObject("page").getString("page_id");
//                if ("good_detail".equals(pageId) || "home".equals(pageId)) {
//                    out.collect(jsonObject);
//                }
//            }
//        });