数据采集的几种方法

数据采集的背景

用户每天通过不同的终端设备在系统A上做出一系列的操作，比如在新闻系统上的看一些文章，进行收藏或者点赞、评论等等，会触发A系统下不同的微服务，如abc等，如何记录好用户的行为信息，并将其保存下来用于分析用户的行为偏好，需要结合不同的数据采集策略。

业务系统

数据采集的几种方式

方式一：读备份库

为了不影响业务系统的正常运行，可以采用读备份库的数据，这样能够及时获取数据进行一些分析工作，但是有些从业务也会读取备份数据库，还需要考虑一致性和可用性问题。

数据库备份

方式一：埋点(pingback)

可以在前端APP上记录用户点击，滑动速度，停留时间，进入的时间段，最后看的新闻等等信息，这些可以通过网络传输将埋点信息记录下来，用于数据分析。但是这种方式有可能会对业务系统代码具有一定的侵入性，同时工作量也比较大，存在一定的安全隐患。

埋点

后端采集数据的service

/**
     * 埋点接收数据
     * @param pingBack
     * @return
     */
    @RequestMapping(path = "/insert", method = RequestMethod.POST)
    @ResponseBody
    public ApiResponse<Boolean> insert(@RequestBody PingBack pingBack) {
        Boolean result = patientService.savePingBack(pingBack);
        return new ApiResponse<Boolean>().success(result);
    }

已有的业务系统可以给数据采集系统发送数据

/**
     * pingback方式插入
     * @param patient
     * @return
     */
    @RequestMapping(path = "/insert", method = RequestMethod.POST)
    @ResponseBody
    public ApiResponse<Boolean> insert(@RequestBody Patient patient) {
        try{
            Boolean result = patientService.savePatient(patient);
            return new ApiResponse<Boolean>().success(result);
        }catch (InternalError error){
            log.error("insert error");
        }finally {
            pingBackService.jsonRequest(url+"insert", patient);
        }
        return null;
    }

方式三：发送消息的方式

上述埋点的方式在业务系统繁忙的情况下，会对数据采集系统产生大量的请求，如果数据处理不及时会把数据采集服务打垮，同时为了解耦，这里可以引入消息中间件，如果对时效性要求较高，可以采用推模式对数据采集系统进行推送，如果时效性不是很高，可以采用定时任务拉取数据，再进行分析。
同时可以多个系统订阅消息中间件中不同Topic的数据，可以对数据进行重用，后端多个数据分析系统之间互不影响，减轻了从业务系统采集多份数据的压力。

引入消息中间件

数据采集Service

/**
    * 消息中间件的方式更新
    * @param patient
    * @return
    */
   @RequestMapping(path = "/update", method = RequestMethod.POST)
   @ResponseBody
   public ApiResponse<Boolean> update(@RequestBody Patient patient) {
       try{
           Boolean result = patientService.updatePatient(patient);
           return new ApiResponse<Boolean>().success(result);
       }catch (InternalError error){
           log.error("update error");
       }finally {
           sendMessageService.send(patient);
       }
       return null;
   }

中间件发送数据实现（以kafka为例）

@Service
@Slf4j
public class SendMessageService {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    @Value("topic")
    private String topic;

    private ObjectMapper om = new ObjectMapper();

    public boolean send(Object object){
        String objectJson = "";
        try {
            objectJson = new ObjectMapper().writeValueAsString(object);
        } catch (Exception e) {
            log.error("can't trans the {} object to json string!", object);
            return false;
        }
        try{
            String result = kafkaTemplate.send("mysql-kafka-patient", objectJson).get().toString();
            if(result!=null){
                return true;
            }

        }catch (Exception e){
            return false;
        }
        return false;
    }
}

中间件拉取数据：

@KafkaListener(id = "forward", topics = "mysql-kafka-patient")
    public String forward(String data) {
        log.info("mysql-kafka-patient "+data);
        JSONObject jsonObject1 = JSONObject.parseObject(data);
        Message message = (Message) JSONObject.toJavaObject(jsonObject1,Message.class);
        messageService.updateMessage(message);
        return data;
    }

方式四：读取MySQL中的binlog

MySQL会把数据的变更（插入和更新）保存在binlog中，需要在my.ini中配置开启，因此采用kafka订阅binlog，会将DB中需要的字段抓取出来，保存在备份库中，进行数据分析，工作量较小，安全稳定。

name=mysql-b-source-pingBack
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:mysql://localhost:3306/test?user=root&password=root
# timestamp+incrementing 时间戳自增混合模式
mode=timestamp+incrementing
# 自增字段  id
timestamp.column.name=commenttime
incrementing.column.name=id
# 白名单表  pingBack
table.whitelist=pingBack
# topic前缀   mysql-kafka-
topic.prefix=mysql-kafka-

具体使用可以参考：官方文档

分析对比

数据采集方式	优点	缺点
埋点(pingback)	很细致的将前端用户操作记录下来，能够感知到DB存储之外的用户信息，时效性高	工作量大，可能对业务代码有侵入性；当业务量大的时候，数据抓取服务也需要承载一定的压力，对数据不方便统计和聚合
主库写备库读	及时感知备库中的信息，数据一致性强	可能存在大量不需要进行分析的字段，对业务性能有影响
埋点+消息中间件	有效的解决业务量大时对数据存取性能的要求，根据数据抓取服务的需求可以拉也可以推，解耦业务代码	可能会丢失数据，降低了时效性
订阅binlog	工作量小，可以离线感知数据的变化，对数据变更进行统计分析	不能感知除DB数据变更之外的用户行为

因此对数据的抓取可以多种方式结合，具体还是要根据后端数据分析任务对数据的时效性、需求和性能综合考虑。

源码分享

Demo1：患者服务（patient）会产生不同的病例记录，然后doctor服务端采用kafka/pingback/binlog三种不同方式感知患者的患病信息，对数据进行有效的采集。
源码参考
Demo2: 一个数据采集的微服务，用户发送请求到微服务系统中，系统通过日志的形式将请求信息持久化到文件中。
源码参考