【Flink实战】新老用户分析：按照操作系统维度进行新老用户的分析

大数据小禅

发布于 2023-09-14 08:10:09

1440

发布于 2023-09-14 08:10:09

日志的数据格式

{
"deviceType":"iPhone 10",
"uid":"user_1",
"product":{
"name":"宝马",
"category":"车"
},
"os":"iOS",
"ip":"171.11.85.21",
"nu":1,
"channel":"华为商城",
"time":1735419335423,
"event":"browse",
"net":"WiFi",
"device":"4759947c-cd47-433c-ac8f-ae923a6d38b6",
"version":"V1.2.0"
}

需求：按照操作系统维度进行新老用户的分析

关键字：操作系统 OS 老用户nu

维度先从单一的开始扩展:操作系统省份的维度

写入数据到Redis 官方文档https://bahir.apache.org/docs/flink/current/flink-streaming-redis/

    <dependency>
        <groupId>org.apache.bahir</groupId>
        <artifactId>flink-connector-redis_2.11</artifactId>
        <version>1.0</version>
    </dependency>

代码

public class OsUserCntAppV1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream = environment.readTextFile("data/access.json");
        environment.setParallelism(1); //设置并行度为1方便观察
        SingleOutputStreamOperator<Access> filter = stream.map(new MapFunction<String, Access>() {
            @Override
            public Access map(String s) throws Exception {
                // json 转 Access
                try {
                    return JSON.parseObject(s, Access.class);
                } catch (Exception e) {
                    e.printStackTrace();
                    return null;
                }
            }
            //这里是只要不为空的数据  x != null等于把上面的空的数据过滤掉
        }).filter(x -> x != null).filter(new FilterFunction<Access>() {
            @Override
            public boolean filter(Access access) throws Exception {
                //只过滤出来 event='startup'的数据
                return "startup".equals(access.event);
            }
        });
        //Access{device='6d27244c-b5e5-4520-9c6d-c4e17e2391fe', deviceType='iPhone 9', os='iOS', event='startup', net='4G', channel='华为商城', uid='user_36', nu=0, nu2=0, ip='123.232.241.201', time=1735419335573, version='V1.2.0', province='null', city='null', product=null}
        // 操作系统维度  新老用户
        //返回三元组 (操作系统 新老用户 数字1)
        SingleOutputStreamOperator<Tuple3<String, Integer, Integer>> sum = filter.map(new MapFunction<Access, Tuple3<String, Integer, Integer>>() {
            @Override
            public Tuple3<String, Integer, Integer> map(Access access) throws Exception {
                return Tuple3.of(access.os, access.nu, 1);
            }
            //根据OS操作系统 与 字段nu新老用户分组  传入三元组返回二元组  
            //因为是按照Tuple2<String, Integer>这两个进行keyby 所以进入三元组出去二元组
        }).keyBy(new KeySelector<Tuple3<String, Integer, Integer>, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> getKey(Tuple3<String, Integer, Integer> value) throws Exception {
                return Tuple2.of(value.f0, value.f1);
            }
            /**
             * (Android,1,1)
             * (iOS,1,1)
             * (iOS,0,1)
             */
            //根据第三个字段进行聚合
        }).sum(2);
        //结果
        /**
         * (Android,1,23)
         * (Android,0,13)
         * (iOS,0,20)
         */

        //单机redis  出现连接超时->修改redis.conf配置文件 bing IP
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("192.168.192.100").build();

        sum.addSink(new RedisSink<>(conf,new RedisSinkForV1()));
        environment.execute("OsUserCntAppV1");
    }
}