spark之Markov马尔可夫智能邮件预测

一.来源

　　此项目来源《Data Algorithms Recipes for Scaling Up with Hadoop and Spark》第11章，本程序利用spark3.0以及java8进行改写，

改写的有:

　　1.利用spark3.0与java8

　　2.直接利用spark生成最终的状态转移矩阵，不用分开统计

　　3.利用python加载状态转移矩阵进行预测

二.目的

　　用户的购买行为看起来是没有规律可循的，但其实从时间有序的角度看，也许是有规律可循的，例如，用户可能每一个月发工资时购买得多，每年某个时间（双十一、生日）等购买得比较多马尔科夫模型能够挖掘出时间上的规律，假设我们能够根据用户上一次购买记录推测其下一次购买时间，就可以在推测时间向其发送邮件进行营销至于营销的商品内容，可以根据其他推荐算法的结果。

三.程序

完整程序见，Markov马尔可夫智能邮件预测(https://github.com/jiangnanboy/spark_tutorial)

1.利用spark统计状态转移矩阵

    /**
     * 建立状态转移概率矩阵
     * @param session
     */
    public static void buildStateTransitionMatrix(SparkSession session, Broadcast<Map<String, String>> broadcastStatesMap, Broadcast<List<Tuple2<String, List<Double>>>> broadcastInitStateList) {
        //customerID,transactionID,purchaseDate,amount（顾客ID,交易ID，交易日期，金额）
        String path = PropertiesReader.get("intermediate_smart_email_txt");
        JavaRDD<String> javaRDD = session.read().textFile(path).toJavaRDD().coalesce(10);

        //key=customerID,v=(purchaseDate,amount)
        JavaPairRDD<String, Tuple2<Long, Integer>> javaPairRDD = javaRDD.mapToPair(line -> {
            String[] tokens = StringUtils.split(line, ",");
            if(4 != tokens.length) {
                return null;
            }
            long date = DateUtils.parseDate(tokens[2], "yyyy-MM-dd").getTime();
            int amount = Integer.parseInt(tokens[3]);
            Tuple2<Long, Integer> t2 = new Tuple2<>(date, amount);
            return new Tuple2<>(tokens[0], t2);
        });

        //group by customerID
        JavaPairRDD<String, Iterable<Tuple2<Long, Integer>>> customerRDD = javaPairRDD.groupByKey();

        //创建状态序列
        JavaPairRDD<String, List<String>> stateSequence = customerRDD.mapValues(dateAndAmount -> {
            List<Tuple2<Long, Integer>> list = toList(dateAndAmount);
            Collections.sort(list, TupleComparatorAscending.INSTANCE);//对list按日期排序
            return toStateSequence(list);
        });

        /**
         * customerID, List<State>
         * 所有状态的频率为1 =》（（fromState, toState）,1）
         *   | S1   S2   S3   ...
         *---+-----------------------
         *S1 | <probability-value>
         *   |
         *S2 |
         *   |
         *S3 |
         *   |
         *...|
         */
        JavaPairRDD<Tuple2<String, String>, Integer> model = stateSequence.flatMapToPair(s -> {
            List<String> states = s._2;
            List<Tuple2<Tuple2<String, String>, Integer>> mapOut = new ArrayList<>();
            if((null == states) || (states.size() < 2)) {
                return Collections.emptyIterator();
            }
            for(int i = 0; i < (states.size() - 1); i++) {
                String fromState = states.get(i);
                String toState = states.get(i+1);
                Tuple2<String, String> t2 = new Tuple2<>(fromState, toState);
                mapOut.add(new Tuple2<>(t2, 1));
            }
            return mapOut.iterator();
        });

        // 统计所有状态频率：  ((fromState, toState), frequence)
        JavaPairRDD<Tuple2<String, String>, Integer> fromStateToStateFrequence1 = model.reduceByKey((i1, i2) -> i1 + i2);

        // ((fromState, toState), frequence) =》 (fromState, (toState, frequence))
        JavaPairRDD<String, Tuple2<String, Integer>> fromStateToStateFrequence2 = fromStateToStateFrequence1.mapToPair(s -> {
            String key = s._1._1;
            Tuple2<String, Integer> value = new Tuple2<>(s._1._2, s._2);
            return new Tuple2<>(key, value);
        });

        // group by fromState =》 fromState，List<Tuple2<toState, frequence>> => rowNumber,List<Tuple2<toState, frequence>>
        JavaPairRDD<String, Iterable<Tuple2<String, Integer>>> groupState = fromStateToStateFrequence2.groupByKey().mapToPair(st2 -> {
            String rowNumber = broadcastStatesMap.getValue().get(st2._1);
            return new Tuple2<>(rowNumber, st2._2);
        });

        //初始化矩阵状态，value = 1.0 / size
        //List<Tuple2<String, List<Double>>> initStateList = initState(broadcastStatesMap.getValue().size());
        JavaPairRDD<String, List<Double>> initStatePairRDD = JavaSparkContext.fromSparkContext(session.sparkContext()).parallelizePairs(broadcastInitStateList.getValue());

        //initStatePairRDD.leftOuterJoin(groupState)
        JavaPairRDD<String, Tuple2<List<Double>, Optional<Iterable<Tuple2<String, Integer>>>>> joinPairRDD = initStatePairRDD.leftOuterJoin(groupState);

        //规范化转移矩阵，使行的概率和为“1”
        JavaPairRDD<String, List<Double>> resultJavaPairRDD = joinPairRDD.mapValues(lot2 -> {
            int size = broadcastStatesMap.getValue().size();
            List<Double> listDouble = lot2._1;
            Optional<Iterable<Tuple2<String, Integer>>> option = lot2._2;
            if(option.isPresent()) {
                Iterable<Tuple2<String, Integer>> toStateFrequence = option.get();
                Iterator<Tuple2<String, Integer>> iter = toStateFrequence.iterator();
                List<Tuple2<String, Integer>> iterList = new ArrayList<>();
                int sum = 0;
                while(iter.hasNext()) {
                    Tuple2<String, Integer> t2 = iter.next();
                    iterList.add(t2);
                    sum += t2._2;
                }
                //加入平滑，防止概率为0
                if(iterList.size() < size) {
                    sum += size;
                    for(int i = 0; i < listDouble.size(); i ++) {
                        listDouble.set(i, 1.0/sum);
                    }
                }

                for(int i = 0; i < iterList.size(); i++) {
                    String stateNumber = broadcastStatesMap.getValue().get(iterList.get(i)._1);
                    double numalizeValue = iterList.get(i)._2 / (double)sum;
                    listDouble.set(Integer.parseInt(stateNumber), numalizeValue);
                }

            } else {
                return listDouble;
            }
            return listDouble;
        });

        //1.利用sortByKey对转移状态排序，最终的状态转移概率矩阵
        //List<Tuple2<String, List<Double>>> stateResult = resultJavaPairRDD.sortByKey().collect();

        //2.利用takeOrdered对转移状态排序，最终的状态转移概率矩阵
        List<Tuple2<String, List<Double>>> stateResult = resultJavaPairRDD.takeOrdered(broadcastStatesMap.getValue().size(), StateTupleComparatorAscending.INSTANCE);

        //打印转移概率矩阵
        for(Tuple2<String, List<Double>> s : stateResult) {
            StringBuilder sb = new StringBuilder();
            sb.append(s._1).append(",");
            for(int i = 0; i < (s._2.size() - 1); i ++) {
                sb.append(s._2.get(i)).append(" ");
            }
            sb.append(s._2.get(s._2.size() - 1));
            System.out.println(sb.toString());
        }

    }

2.利用python加载状态转移矩阵并进行预测

import os
import time
import datetime

# 根据（spark）Markov.java统计出的马尔可夫模型(model.txt)，对validate.txt中的数据进行预测什么时间应该发出营销邮件
user_action = {}
model = []
#9大状态
states = ["SL", "SE", "SG", "ML", "ME", "MG", "LL", "LE", "LG"]

validate_path = os.path.join(os.getcwd(), "validate.txt")
model_path = os.path.join(os.getcwd(), "model.txt")

#读取validate data
with open(validate_path, 'r', encoding='utf-8') as f_read:
    for line in f_read:
        items = line.strip().split(',')
        user_id = items[0]
        if user_id in user_action.keys():
            hist = user_action[user_id]
            lst = [items[2], items[-1]]
            hist.append(lst)
        else:
            hist = []
            hist.append([items[2], items[-1]])
            user_action[user_id] = hist
print(user_action)

#读取model data
with open(model_path, 'r', encoding='utf-8') as f_read:
    for line in f_read:
        items = line.strip().split()
        row = []
        for item in items:
            row.append(float(item))
        model.append(row)
print(model)

#根据最近客户的行为数据(至少两次交易)make prediciton
for user_id,user_action_list in user_action.items():
    if len(user_action_list) < 2:
        continue
    state_sequence = []
    last_date = ''
    prior = user_action_list[0]
    for i in range(1, len(user_action_list)):
        current = user_action_list[i]
        prior_date = prior[0]
        current_date = current[0]

        #相隔天数
        prior_date = time.strptime(prior_date, '%Y-%m-%d')
        current_date = time.strptime(current_date, '%Y-%m-%d')
        prior_date = datetime.datetime(prior_date[0], prior_date[1], prior_date[2])
        current_date = datetime.datetime(current_date[0], current_date[1], current_date[2])
        days_diff = (current_date - prior_date).days

        dd = 'L'
        if days_diff < 30:
            dd = 'S'
        elif days_diff < 60:
            dd = 'M'

        #相差金额
        prior_amount = int(prior[1])
        current_amount = int(current[1])

        ad = 'G'
        if prior_amount < 0.9 * current_amount:
            ad = 'L'
        elif prior_amount < 1.1 * current_amount:
            ad = 'E'

        state_sequence.append(dd+ad)

        prior = current
        last_date = current_date

    if state_sequence:
        #根据最近一个状态发送营销邮件日期
        last_state = state_sequence[-1]
        row_index = states.index(last_state)
        row_value = model[row_index] #转移矩阵中行号为row_index的这一行值
        max_value = max(row_value) #row_value中最大值
        col_index = row_value.index(max_value) #max_value的索引号
        next_state = states[col_index]

        if next_state.startswith('S'):
            next_date = last_date + datetime.timedelta(15)
        elif next_state.startswith('E'):
            next_date = last_date + datetime.timedelta(45)
        else:
            next_date = last_date + datetime.timedelta(90)

    print('用户：{}, 预测下次邮件发送时间：{}'.format(user_id, next_date))

 3.状态转移矩阵

4.396976638863118E-6 4.396976638863118E-6 0.8062208425486636 0.15858575643387607 4.396976638863118E-6 4.396976638863118E-6 0.035153828227710626 4.396976638863118E-6 4.396976638863118E-6
0.1111111111111111 0.1111111111111111 0.1111111111111111 0.1111111111111111 0.1111111111111111 0.1111111111111111 0.1111111111111111 0.1111111111111111 0.1111111111111111
0.8043794973142799 3.1671227323401235E-6 3.1671227323401235E-6 0.1594804651869869 3.1671227323401235E-6 3.1671227323401235E-6 0.03611153339414209 3.1671227323401235E-6 3.1671227323401235E-6
1.225940909648155E-5 1.225940909648155E-5 0.8156797842343999 1.225940909648155E-5 1.225940909648155E-5 0.15299742552408974 0.031212455559642024 1.225940909648155E-5 1.225940909648155E-5
0.010869565217391304 0.010869565217391304 0.7282608695652174 0.010869565217391304 0.010869565217391304 0.11956521739130435 0.05434782608695652 0.010869565217391304 0.010869565217391304
4.122521334047904E-5 4.122521334047904E-5 0.8190625386486375 0.14758626375891495 4.122521334047904E-5 4.122521334047904E-5 0.03298017067238323 4.122521334047904E-5 4.122521334047904E-5
4.703226413319537E-5 4.703226413319537E-5 0.8348697206283511 4.703226413319537E-5 4.703226413319537E-5 0.14081459881478695 4.703226413319537E-5 4.703226413319537E-5 0.023845357915530052
0.017857142857142856 0.017857142857142856 0.6785714285714286 0.017857142857142856 0.017857142857142856 0.125 0.017857142857142856 0.017857142857142856 0.03571428571428571
5.083884087442806E-4 5.083884087442806E-4 0.7961362480935434 5.083884087442806E-4 5.083884087442806E-4 0.16573462125063548 0.03304524656837824 5.083884087442806E-4 5.083884087442806E-4

 4.预测结果

用户：user1, 预测下次邮件发送时间：2020-01-22 00:00:00
用户：user2, 预测下次邮件发送时间：2020-02-16 00:00:00
用户：user3, 预测下次邮件发送时间：2020-01-25 00:00:00
用户：user4, 预测下次邮件发送时间：2020-04-05 00:00:00
用户：user5, 预测下次邮件发送时间：2020-01-30 00:00:00
用户：user6, 预测下次邮件发送时间：2020-01-16 00:00:00