Yearly Archives: 2015

Sqoop的Mysql数据导出实现分两种，一种是使用JDBC方式从Mysql中获取数据，一种是使用MysqlDump命令从MySql中获取数据，默认是 JDBC方式获取数据，如果要使用dump方式获取数据，需要添加 -direct 参数。 使用JDBC方式从Mysql中获取数据 配置语句时，需要添加 $CONDITIONS 点位符，比如：SELECT id FROM user WHERE $CONDITIONS，Sqoop在内部实现时会把它替换成需要的查询条件。 Sqoop启动后会先查询元数据，它会把 $CONDITIONS 替换为 (1=0) ，然后用得到的SQL语句查询数据表对应的Meta信息对于导出一个表的情况，Sqoop会使用这个SQL查询三次数据库，分别是： 1、获取 colInfo（最终得到columnTypes信息）2、查询ColumnNames信息3、生成QueryResult类执行 generateFields操作获取columnTypeNames时。 Sqoop会对获取的Fields做校验，列不能重复，它还会处理数据库的字段到Java属性名的转换 QueryResult类是通过构建java类文件，然后获取JavaCompiler，然后编译加载，为了提高处理性能，不是使用反射实现的，这个生成类内部处理mysql到hdfs属性值为空和分隔符的处理。 接着它会进行下面一个Sql查询操作，查询结果集为MIN(split列),MAX(split列)，查询条件的处理逻辑为 $CONDITIONS 替换为(1=1)，然后再添加外面SELECT查询 (举例：SELECT MIN(id), MAX(id) FROM (SELECT ID,NAME,PASSPORT WHERE (1=1) ) AS t1 )，这样就查询出来此次导出数据最大的split列值和最小的split列值。 对于为整数、布尔值、时间格式、Float等 的分区列，进行split时直接根据对应值的大小进行Split，Text文本的处理方式比较特殊，Sqoop先会对之前获取到的Min和Max的字串寻找它们最大的相同前缀子字串，然后把后面的字段转化为BigDecimal，结合时char占两个字节（65536），算法在 … Continue reading →

协议： 输出文件压缩：Camus默认只支持两种压缩格式(snappy和deflate)，默认是defalte，使用 StringRecordWriterProvider写入文本格式文档时，还可以指定gzip的压缩格式，扩展其它压缩格式很容易，只需要添加 两行代码就可以，建议增加lzo和lzop的压缩格式，以和我们Hive保持一致。 输出格文件类型：建议文本格式的文件 文件目录规则：（配置的目录）+ topic名 + daily|hour + （年/月/日）|（年/月/日/小时） + 数据文件，例如：/rocketmq/data/vip_ods_heartbeat/daily/2015/06/10 /vip_ods_heartbeat.broker-a.0.999.48388735.1433865600000.deflate 文件名规则：topic名+ （RocketBrokerId）|（kafka的对应分区的learder的BrokerId）+ （RocketQueueId）|（kafka分区号）+ 写入消息行数 + 最后一条消息的Offset + 编码的分区（时间 + 压缩格式后缀）,例如：vip_ods_heartbeat.broker- a.0.999.48388735.1433865600000.deflate Topic的命名规则：业务标识+数据库名+数据库表名（分表只需要BaseName就可以），例 如：vip_ods_heartbeat 消息格式：操作类型\t表名（分表的话是分表名）\t数据库名\t主键名\t唯一索引\tBinlog日志时间 \tCheckPoint字段\tDataBefore\tDataAfter，库名表名都是RockMQ中的原始数据，在生成列数据时，列中数 据如果有\t等特殊字符需要替换，例如：insert\theartbeat\tvip_ods\tid\tname,pid\t1232132131\t120@21\t{“字段名”,”字段 值”,…}\t{“字段名”,”字段值”,…} 确定数据导入是否完成： Camus中会在History的目录中存放历次消费的状态，包括开始执行的分区和它们的Offset、 执行结束位置的分区和它们的Offset，这两个文件以SequenceFile的形式存放在HDFS文件中 Camus在执行结束后可以把 执行信息汇总发送到Kafka的Topic中，Topic的名字为：TrackingMonitoringEvent，如果监控程序监控这个 Topic，是可以得到当前执行的情况的信息的。