mysql同步数据到hive---binlog方式

mysql同步数据到hive大部分公司目前都是走的jdbc的方式。 这种方式有两个好处： 也有不好的地方： 这一步最主要的细节是将mysql库的所有binlog数据全部打入一个kafka topic，格式使用json。格式如下： 这一步的主要的细节在于写入到hdfs的结构，以及为什么不直接写入hive。 不写入到hive表的原因在于，binlog的数据结构是不固定的，而hive的结构相对是比较固定的。如果要写入到hive的话，就需要将不同的表的binlog写入到不同的hive表中，这个维护成本太高了。而且spark其实可以直接读取hdfs的json文件，因此直接放hdfs就好了。 写入到hdfs的话，考虑到后续读这个数据是要按照表去读增量数据，所以写入的目录一定是要带日期和表名称的。我这边用的目录结构是这样的： 也就是说要在flink根据数据所属的db、table_name、和日期将数据写入到不同的目录里。 在这一步的处理的过程中遇到了一些比较重要的参数问题。 2.如上所述checkpoint的时间间隔。不仅仅会影响checkpoint的频率，而且会影响hdfs文件的大小，而hdfs文件的大小可能会对hdfs的性能有很大影响。这个值如果太大，就会造成数据延迟太高，如果太小就会造成小文件过多。我这边设置的是5分钟。 细心的看官，这个时候会问了，既然你的目录是分table的，那么每个table每5分钟的binlog数据量是不一样的。对于某些大的mysql表，我们可能每5分钟生成一个文件还能接受。对于一些比较小的表，每五分钟生成一个文件那么文件就会非常小。所以我这边又做了一层的筛选，我把mysql的大的表筛选出来，只同步大的表到hdfs，用以binlog的数据同步。因为本身binlog的方式同步mysql数据为的就是节约mysql的读取压力，而小的表对于不会有太大压力，这些表可以直接通过jdbc的方式去同步。 这个是整个环节里面最复杂的一部分，涉及的细节也比较多。 首先，我们要明确一下总体的思路是什么。总体的思路就是要读取hdfs上的老的历史数据，然后和新的binlog数据合并生成新的快照。 其实这中间还涉及到一些其他的细节，比如mysql表结构变更，或者mysql和hive的数据结构不一致的情况。 另外我们这边还存在多个db的相同的表导入到hive的一张表中的其他问题，我就不赘述了。

通过Map/Reduce进行批处理递送到Apache

Hadoop仍然是中枢环节。,但随着要从“超思维速度“分析方面获取竞争优势的压力递增，因此Hadoop(分布式文件系统)自身经历重大的发展。科技的发展允许实时查询，如Apache

Drill,Cloudera

Impala和Stinger

Initiative正脱颖而出，新一代的资源管理Apache

YARN

支持这些。

为了支持这种日渐强调实时性 *** 作,发布一个新MySQL

Applier

for

Hadoop(用于Hadoop的MySQL

Applier)组件。它能够把MySQL中变化的事务复制到Hadoop

/

Hive

/

HDFS。Applier

组件补充现有基于批处理Apache

Sqoop的连接性。

这个组件(MySQL

Applier

for

Hadoop)的复制是通过连接MySQL主服务，一旦二进制日志被提交，就读取二进制日志事务，并且把它们写到HDFS.

这个组件使用libhdfs提供的API，一个C库 *** 作HDFS中的文件。这库由Hadoop版本预编译生成的。

它连接MySQL主服务读二进制日志，然后提取发生在主服务上的行插入事件，解码事件,提取插入到行的每个字段的数据,并使用满意的处理程序得到被要求的格式数据。把它追加到HDFS

中一个文本文件。

数据库被映射为单独的目录,它们的表映射为子目录，保存在数据仓库目录。每个表的数据被写到Hive/

HDFS中文本文件(称为datafile1.txt)。数据可以用逗号格式分隔或其他格式,那可用命令行参数来配置的。

1、在MYSQL中创建一个参数表A，用来记录要传输的表的表名B和字段。 2、HDFS上的数据的格式和MYSQL中参数表A中B表的字段的数量一样。 3、从MYSQL的参数表A中，读取要传输的表B。把这些字段拼凑成一个insert语句，例如 ，拼凑为insert into b

---