Spark解析binlog日志，写入MySQL

1. 背景

  由于公司业务线的不断拓展，创建了很多MySQL实例，为了安全起见每个实例之间不能直接互相访问，但是业务部门又需要整合各个业务线的数据进行分析、制定风控策略等。因此需要将不同业务线数据进行归集。

  当然一下方案不是最优的，MySQL实例之间数据互通，有很多成熟且稳定的方式，因此我觉得我们选择了一种不是非常理想的方式！

2.  处理流程

    MySQL ——> 产生binlog日志 ——> Maxwell解析成json格式 ——> 发送到Kafka ——> 通过Spark消费 ——> 写入MySQL

3.  处理思路

  由于需要同步的表非常多，因此通过JavaBean反射的方式去解析Maxwell发送到kafka的数据工作量较繁重。并且对于不断变更表结构的业务来讲，也不是一个非常合理的方式。

  Maxwell针对不同的MySQL操作方式会产生不同结构的Json数据：

mysql> insert into `test`.`maxwell` set id = 1, daemon = 'Stanislaw Lem';
  maxwell: {
    "database": "test",
    "table": "maxwell",
    "type": "insert",
    "ts": 1449786310,
    "xid": 940752,
    "commit": true,
    "data": { "id":1, "daemon": "Stanislaw Lem" }
  }
           

mysql> update test.maxwell set daemon = 'firebus!  firebus!' where id = 1;
  maxwell: {
    "database": "test",
    "table": "maxwell",
    "type": "update",
    "ts": 1449786341,
    "xid": 940786,
    "commit": true,
    "data": {"id":1, "daemon": "Firebus!  Firebus!"},
    "old":  {"daemon": "Stanislaw Lem"}
  }
           

  经过分析我们可以解析出相应的SQL语句，然后通过JDBC的方式操作MySQL，实现跨库的数据同步需求。

4、遇到问题

  如果一条数据中存在为NULL值的字段，Maxwell不会将该字段通过Json的方式发送过来，因此Json中不完全包含一条完整的数据，该字段在操作MySQL时只能将其插入对应字段的默认值，但不影响数据的准确性。

  当一条数据中某个字段由NULL变更为非NULL值时，我们试图通过解析old串中被更新的字段时是徒劳的，因为此时old串为空，此时会报：java.lang.ArrayIndexOutOfBoundsException: -1（数组越界异常）。

  当一条数据中某个字段由非NULL变更为NULL值时，在data串中你也是无法获取到该字段的，试图解析会出现：java.lang.NullPointerException（空指针异常）。

  解决上述两种问题的方式是对update语句分情况解析。如果被变更状态的字段都出现在了data串中，那么可以将data串作为最新的数据更新到MySQL中。否则通过比较data和old串的不同，还原相应字段的最新值，从而更新MySQL，最终问题得到解决。