背景: hive中虽然有数据类型,但是int、double、string都可以在hive中存储为string类型,hive中String类型是没有长度限制的。hive中空值有三种情况,null(NULL值)、'\0'字符
(hive自定义的空字符)、''(空字符)。
由于hive在存储上存在以上情况,当把数据导入mysql会导致一下问题:
1、字符长度太小的异常。由于mysql定义表结构的时候会定义字段长度值,当hive中数据的int、double、String长度大于mysql表定义的长度会出现mysql字段定义的数据长度太多而出现异常。
2、int、double空数据异常。hive中int、double的空字符是有以上三种情况,会导致空值存储到mysql时出现类型不匹配而出现异常。
解决办法:
1、根据业务场景设定mysql字段长度。
2、在数据导从hive导入mysql之前把空字符替换成默认字符。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的一张表中的其他问题,我就不赘述了。直接导入hive表
sqoop import --connect jdbc:postgresql://ip/db_name--username user_name --table table_name --hive-import -m 5
内部执行实际分三部,1.将数据导入hdfs(可在hdfs上找到相应目录),2.创建hive表名相同的表,3,将hdfs上数据传入hive表中
sqoop根据postgresql表创建hive表
sqoop create-hive-table --connect jdbc:postgresql://ip/db_name --username user_name --table table_name --hive-table hive_table_name( --hive-partition-key partition_name若需要分区则加入分区名称)
导入hive已经创建好的表中
sqoop import --connect jdbc:postgresql://ip/db_name --username user_name --table table_name --hive-import -m 5 --hive-table hive_table_name (--hive-partition-key partition_name --hive-partition-value partititon_value)
使用query导入hive表
sqoop import --connect jdbc:postgresql://ip/db_name --username user_name --query "select ,* from retail_tb_order where \$CONDITIONS" --hive-import -m 5 --hive-table hive_table_name (--hive-partition-key partition_name --hive-partition-value partititon_value)
注意:$CONDITIONS条件必须有,query子句若用双引号,则$CONDITIONS需要使用\转义,若使用单引号,则不需要转义。
评论列表(0条)