搜索[实践]结果-低调大师优秀个人博客

MySQL表碎片整理

ALTER TABLE 使用 innodb_file_per_table 可以更好地管理空间备份：重要操作前建议备份数据复制环境：在主从复制环境中，OPTIMIZE TABLE 会被复制到从库最佳实践

2025-10-26

数据通信数据通信整体的类如下，MarshallingCodeCFactory可以省去，这个只是代替java的序列化的功能，因为java的序列化功能效率低： Client.java package bhz.netty.runtime; import io.netty.bootstrap.Bootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import io.netty.handler.timeout.ReadTimeoutHandler; import java.util.concurrent.TimeUnit; /** * Best Do It */ public class Client { private static class SingletonHolder { static final Client instance = new Client(); } public static Client getInstance(){ return SingletonHolder.instance; } private EventLoopGroup group; private Bootstrap b; private ChannelFuture cf ; private Client(){ group = new NioEventLoopGroup(); b = new Bootstrap(); b.group(group) .channel(NioSocketChannel.class) .handler(new LoggingHandler(LogLevel.INFO)) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder()); sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder()); //超时handler（当服务器端与客户端在指定时间以上没有任何进行通信，则会关闭响应的通道，主要为减小服务端资源占用） sc.pipeline().addLast(new ReadTimeoutHandler(5)); sc.pipeline().addLast(new ClientHandler()); } }); } public void connect(){ try { this.cf = b.connect("127.0.0.1", 8765).sync(); System.out.println("远程服务器已经连接, 可以进行数据交换.."); } catch (Exception e) { e.printStackTrace(); } } public ChannelFuture getChannelFuture(){ if(this.cf == null){ this.connect(); } if(!this.cf.channel().isActive()){ this.connect(); } return this.cf; } public static void main(String[] args) throws Exception{ final Client c = Client.getInstance(); //c.connect(); ChannelFuture cf = c.getChannelFuture(); for(int i = 1; i <= 3; i++ ){ Request request = new Request(); request.setId("" + i); request.setName("pro" + i); request.setRequestMessage("数据信息" + i); cf.channel().writeAndFlush(request); TimeUnit.SECONDS.sleep(4); } cf.channel().closeFuture().sync(); //前面的主线程已经断开了连接，现在创建一个子线程继续建立连接 new Thread(new Runnable() { @Override public void run() { try { System.out.println("进入子线程..."); ChannelFuture cf = c.getChannelFuture(); System.out.println(cf.channel().isActive()); System.out.println(cf.channel().isOpen()); //再次发送数据 Request request = new Request(); request.setId("" + 4); request.setName("pro" + 4); request.setRequestMessage("数据信息" + 4); cf.channel().writeAndFlush(request); cf.channel().closeFuture().sync(); System.out.println("子线程结束."); } catch (InterruptedException e) { e.printStackTrace(); } } }).start(); System.out.println("断开连接,主线程结束.."); } } ClientHandler.java package bhz.netty.runtime; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import io.netty.util.ReferenceCountUtil; public class ClientHandler extends ChannelHandlerAdapter{ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { Response resp = (Response)msg; System.out.println("Client : " + resp.getId() + ", " + resp.getName() + ", " + resp.getResponseMessage()); } finally { ReferenceCountUtil.release(msg); } } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); } } Server.java package bhz.netty.runtime; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import io.netty.handler.timeout.ReadTimeoutHandler; public class Server { public static void main(String[] args) throws Exception{ EventLoopGroup pGroup = new NioEventLoopGroup(); EventLoopGroup cGroup = new NioEventLoopGroup(); ServerBootstrap b = new ServerBootstrap(); b.group(pGroup, cGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) //设置日志 .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new ChannelInitializer<SocketChannel>() { protected void initChannel(SocketChannel sc) throws Exception { sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder()); sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder()); sc.pipeline().addLast(new ReadTimeoutHandler(5)); sc.pipeline().addLast(new ServerHandler()); } }); ChannelFuture cf = b.bind(8765).sync(); cf.channel().closeFuture().sync(); pGroup.shutdownGracefully(); cGroup.shutdownGracefully(); } } ServerHandler.java package bhz.netty.runtime; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; public class ServerHandler extends ChannelHandlerAdapter{ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { Request request = (Request)msg; System.out.println("Server : " + request.getId() + ", " + request.getName() + ", " + request.getRequestMessage()); Response response = new Response(); response.setId(request.getId()); response.setName("response" + request.getId()); response.setResponseMessage("响应内容" + request.getId()); ctx.writeAndFlush(response);//.addListener(ChannelFutureListener.CLOSE); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); } } Request.java package bhz.netty.runtime; import java.io.Serializable; public class Request implements Serializable{ private static final long SerialVersionUID = 1L; private String id ; private String name ; private String requestMessage ; public String getId() { return id; } public void setId(String id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } public String getRequestMessage() { return requestMessage; } public void setRequestMessage(String requestMessage) { this.requestMessage = requestMessage; } } Response.java package bhz.netty.runtime; import java.io.Serializable; public class Response implements Serializable{ private static final long serialVersionUID = 1L; private String id; private String name; private String responseMessage; public String getId() { return id; } public void setId(String id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } public String getResponseMessage() { return responseMessage; } public void setResponseMessage(String responseMessage) { this.responseMessage = responseMessage; } } 心跳检测心跳检测需要用到sigar.jar（它能读取内存或者cpu等等一些信息），还要用到队列线程池。总的代码如下： Client.java和Server.java几乎没变什么（体现了netty将业务分离的好处），主要是他们两个handler的变化。 ClienHeartBeattHandler.java package bhz.netty.heartBeat; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import io.netty.util.ReferenceCountUtil; import java.net.InetAddress; import java.net.UnknownHostException; import java.util.HashMap; import java.util.Map; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; import org.hyperic.sigar.CpuPerc; import org.hyperic.sigar.Mem; import org.hyperic.sigar.Sigar; import org.hyperic.sigar.Swap; public class ClienHeartBeattHandler extends ChannelHandlerAdapter { private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1); private ScheduledFuture<?> heartBeat; //主动向服务器发送认证信息 private InetAddress addr ; private static final String SUCCESS_KEY = "auth_success_key"; @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { addr = InetAddress.getLocalHost(); String ip = addr.getHostAddress(); String key = "1234"; //证书 String auth = ip + "," + key; ctx.writeAndFlush(auth); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { if(msg instanceof String){ String ret = (String)msg; if(SUCCESS_KEY.equals(ret)){ // 握手成功，主动发送心跳消息 this.heartBeat = this.scheduler.scheduleWithFixedDelay(new HeartBeatTask(ctx), 0, 2, TimeUnit.SECONDS); System.out.println(msg); } else { System.out.println(msg); } } } finally { ReferenceCountUtil.release(msg); } } private class HeartBeatTask implements Runnable { private final ChannelHandlerContext ctx; public HeartBeatTask(final ChannelHandlerContext ctx) { this.ctx = ctx; } @Override public void run() { try { RequestInfo info = new RequestInfo(); //ip info.setIp(addr.getHostAddress()); Sigar sigar = new Sigar(); //cpu prec CpuPerc cpuPerc = sigar.getCpuPerc(); HashMap<String, Object> cpuPercMap = new HashMap<String, Object>(); cpuPercMap.put("combined", cpuPerc.getCombined()); cpuPercMap.put("user", cpuPerc.getUser()); cpuPercMap.put("sys", cpuPerc.getSys()); cpuPercMap.put("wait", cpuPerc.getWait()); cpuPercMap.put("idle", cpuPerc.getIdle()); // memory Mem mem = sigar.getMem(); HashMap<String, Object> memoryMap = new HashMap<String, Object>(); memoryMap.put("total", mem.getTotal() / 1024L); memoryMap.put("used", mem.getUsed() / 1024L); memoryMap.put("free", mem.getFree() / 1024L); info.setCpuPercMap(cpuPercMap); info.setMemoryMap(memoryMap); ctx.writeAndFlush(info); } catch (Exception e) { e.printStackTrace(); } } public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); if (heartBeat != null) { heartBeat.cancel(true); heartBeat = null; } ctx.fireExceptionCaught(cause); } } } ServerHeartBeatHandler.java package bhz.netty.heartBeat; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import java.util.HashMap; public class ServerHeartBeatHandler extends ChannelHandlerAdapter { /** key:ip value:auth */ private static HashMap<String, String> AUTH_IP_MAP = new HashMap<String, String>(); private static final String SUCCESS_KEY = "auth_success_key"; static { AUTH_IP_MAP.put("192.168.1.200", "1234"); } private boolean auth(ChannelHandlerContext ctx, Object msg){ //System.out.println(msg); String [] ret = ((String) msg).split(","); String auth = AUTH_IP_MAP.get(ret[0]); if(auth != null && auth.equals(ret[1])){ ctx.writeAndFlush(SUCCESS_KEY); return true; } else { ctx.writeAndFlush("auth failure !").addListener(ChannelFutureListener.CLOSE); return false; } } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { if(msg instanceof String){ auth(ctx, msg); } else if (msg instanceof RequestInfo) { RequestInfo info = (RequestInfo) msg; System.out.println("--------------------------------------------"); System.out.println("当前主机ip为: " + info.getIp()); System.out.println("当前主机cpu情况: "); HashMap<String, Object> cpu = info.getCpuPercMap(); System.out.println("总使用率: " + cpu.get("combined")); System.out.println("用户使用率: " + cpu.get("user")); System.out.println("系统使用率: " + cpu.get("sys")); System.out.println("等待率: " + cpu.get("wait")); System.out.println("空闲率: " + cpu.get("idle")); System.out.println("当前主机memory情况: "); HashMap<String, Object> memory = info.getMemoryMap(); System.out.println("内存总量: " + memory.get("total")); System.out.println("当前内存使用量: " + memory.get("used")); System.out.println("当前内存剩余量: " + memory.get("free")); System.out.println("--------------------------------------------"); ctx.writeAndFlush("info received!"); } else { ctx.writeAndFlush("connect failure!").addListener(ChannelFutureListener.CLOSE); } } } RequestInfo.java package bhz.netty.heartBeat; import java.io.Serializable; import java.util.HashMap; public class RequestInfo implements Serializable { private String ip ; private HashMap<String, Object> cpuPercMap ; private HashMap<String, Object> memoryMap; //.. other field public String getIp() { return ip; } public void setIp(String ip) { this.ip = ip; } public HashMap<String, Object> getCpuPercMap() { return cpuPercMap; } public void setCpuPercMap(HashMap<String, Object> cpuPercMap) { this.cpuPercMap = cpuPercMap; } public HashMap<String, Object> getMemoryMap() { return memoryMap; } public void setMemoryMap(HashMap<String, Object> memoryMap) { this.memoryMap = memoryMap; } } 文件服务器先略了，貌似用处不大。。。

2018-10-31

PWA实践

PWA实践引言最近在搞基础框架，使用Docker部署到我1M带宽，1核CPU的阿里云上，在演示或者被别人打开的时候老是被吐槽加载太慢了，那么问题来了，怎么优化哪？

2018-10-15

MaxCompute实践

一、写在前面本人之前一直从事程序开发的（PHP、JAVA、Python）工作，在之前的工作经历中有过一段时间配合Hadoop工程师的事务，但接触的并不深，只能说略知点皮毛，有点管中窥豹的感觉。今年进了新公司，因为公司正在组建新的数据部门，非常有幸本人得以调入该部门，恰逢MaxCompute横空出世，刚好因为我那时工作比较空闲，得以安排调研它的功能及测试是否符合我们的要求。由于个人对大数据这块的经验不足，涉及的内容也不是太高端的，所得知识基本是通过阅读文档获得，所以本文仅以初学者的角度去阐述，主要围绕数据迁移这块，其他部分还未来得急涉及，如有不正确的地方，还请指出，谢谢。二、平台体系的选型因初期数据量相对较小，使用Kettle进行抽取数据等工作，ETL的工作大部分在MySQL数据仓库中完成。多种数据源使用Presto（集群）作为查询中间键进行相应的数据分析。但随着业务的疯狂增长，数据表单表达到数亿后，磁盘容量达数几百GB时，数据要求的复杂度逐步提升，使用MySQL作为基础数据仓库的基石已经不足以应付，常出现查询响应时间等待过长，甚至内存崩溃导致执行失败的情况，极大的影响了工作效率。所以使用选择一款大数据平台势在必行。初期考虑使用Hadoop来做进行数据分析平台，恰逢MaxCompute横空出世给我们一个新的选择，在经过仔细比对后，考虑到公司新数据部门刚成立不久，Hadoop大数据相关人才储备较为紧缺，如果等待Hadoop体系完成搭建并使用，可能需要更多的时间，而这势必会拖慢之后的工作进度。而如果使用MaxCompute相对于自建Haoop数据平台要简单，快速的多，但任何产品在正式使用前必然需要进行详细的调研，所以我就开始调研的过程。三、数据迁移 3.1 数据源类型介绍目前我们主要有两种数据源的数据需要放至MaxCompute中进行处理，一种为MySQL，一种为MongoDB。在此主要描述MySQL相关的，至于MongoDB嘛，看到最后你就会知道了^_^。 3.2 迁移工具的选择虽然MaxCompute有多种迁移方案可供选择，但如果使用大数据开发套件上的脚本模式能够有效完成的话，个人还是比较倾向使用大数据开发套件去完成，除非在此之上无法完成。因为相对于自己使用SDK直接同步或其他工具去实现，大数据开发平台要简单快速的多，任务调度系统完善，任务运行情况相对清晰便于查错，之后更多任务可以灵活组合搭配。 3.3 同步策略介绍数据主要分为会变化的数据（有修改+有增加）与不会变化的数据（只增加），官方文档中是建议每天全量的同步策略，即每天的数据作为一个副本存至同步日期为分区，这样做确实有很多的好处。但实际情况可能由于数据量过大，每天同步可能会花更多的时间(测试过1亿数据大概在3~4小时左右，脚本参数配置speed": { "concurrent": "1", "mbps": "1" })，因为我们业务中不存在DELTE操作，所以我们这里处理不管数据是否变化都使用每日增量的方式处理，最终按数据的创建日期存放在对应的分区（前期也不知道怎么设计，如何设计最佳，所以就先设计为1级分区），虽然这样需要多做一步合并操作，多耗费一些资源费用，但确实是实现我们的要求。 3.4 步骤介绍在“大数据开发套件”中同步数据主要使用时间段的方式进行同步，主要分为两个步骤，第一步为手动全量同步，第二步为自动每日增量。如果想让数据的每条数据按数据的创建时间存放到对应的日期的话，则需要在此基础上进行细化。针对不同的数据（会变化与不会变化）的处理方式也各有所不同，基本上是按照文档中所描述的进行了，只是稍微有些不同。 3.5 举例说明 3.5.1 准备工作假设当前日期为： 2017-07-13 Mysql表信息数据表名: test 创建时间字段: created_at更改时间字段: updated_at时间字段值: unix时间戳表结构定义如下： CREATE TABLE `test` ( `uid` int(11) NOT NULL AUTO_INCREMENT COMMENT '用户ID', `name` varchar(10) NOT NULL COMMENT '姓名', `age` tinyint(4) NOT NULL COMMENT '年龄', `sex` tinyint(4) NOT NULL DEFAULT '0' COMMENT '性名(1=男, 2=女, 0=未选择)', `created_at` int(11) NOT NULL COMMENT '创建时间', `updated_at` int(11) NOT NULL COMMENT '修改时间', PRIMARY KEY ( `uid`), KEY `created_at` ( `created_at`), KEY `updated_at` ( `updated_at`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='用户信息表' MaxCompute数据表创建一张与这对应的MaxCompute数据表数据表名: odps_test_history分区名: pt分区值: 数据的创建日期表结构定义如下： CREATE TABLE odps_test_history ( uid BIGINT COMMENT '用户ID', name STRING COMMENT '姓名', age BIGINT COMMENT '年龄', sex BIGINT COMMENT '性名(1=男, 2=女, 0=未选择)', created_at BIGINT COMMENT '创建时间', updated_at BIGINT COMMENT '修改时' ) PARTITIONED BY ( pt STRING ) LIFECYCLE 100000; 命名约定临时表表名后缀：_history 增量表表名后缀：_ inc 3.5.1 同步不会变化的数据 1. 手动全量同步通过脚本模式将小于当前日期(2017-07-13)的所有的数据一次性导入到一张临时表中。在数据开发中点击测试运行，看到任务成功运行后，在任务设置调度参数中设置暂停（测试运行不会受到调度干扰，直接运行）。脚本配置如下： { "configuration": { "reader": { "plugin": "mysql", "parameter": { "datasource": "mysql_db001", "column": [ "uid", "name", "age", "sex", "created_at", "updated_at" ], "where": "created_at < UNIX_TIMESTAMP('20170713')", "splitPk": "uid", "table": "test" } }, "writer": { "plugin": "odps", "parameter": { "partition": "pt=history", "truncate": true, "datasource": "odps_first", "column": [ "uid", "name", "age", "sex", "created_at", "updated_at" ], "table": "odps_test_history" } }, "setting": { "errorLimit": { "record": "0" }, "speed": { "concurrent": "1", "mbps": "1" } } }, "type": "job", "version": "1.0" } 2. 创建最终目标表执行ODPS_SQL复制odps_test_history表并命名为odps_test作为最终目标表。 -- 创建目标表 CREATE TABLE odps_test LIKE odps_test_history; 3. 根据数据的时间进行动态分区执行ODPS_SQL将数据导入到最终目标表中。 --- 将历史表中的数据根据数据的创建的时间插入目标表对应的分区中 INSERT OVERWRITE TABLE odps_test PARTITION (pt) SELECT uid, name, age, sex, created_at, updated_at, TO_CHAR(FROM_UNIXTIME(created_at), 'yyyymmdd') AS pt FROM odps_test_history; 4. 简单验证数据执行ODPS SQL，如两个数一样则说明导入成功，如不一样，请重试。 -- 验证数据总数 SELECT COUNT(*) FROM odps_test_history; SELECT COUNT(*) FROM odps_test; 5. 删除历史临时表如确认数据无误，则可以删除临时表。当然你也可以不删除，以便出了错只需将此时间节点内的数据导入即可，避免从头导入的过程。 --- 删除历史表 ALTER TABLE odps_test_history DROP IF EXISTS PARTITION ( pt='history' ); 6. 更改同步脚本更新同步脚本，并将任务进行提交，之前的第一步操作的暂停记得勾掉。更改后的同步脚本如下，请注意蓝色部分： 7. 完成如果执行到这一步基本上就完成了，隔天就会在你在设定时间内运行，你也可以去运维中心中查看你的任务情况。下面附上”不会变化的数据“数据同步的全部流程图： 3.5.2 同步会变化的数据注意事项：A. 为方便演示，继续使用test表来作为演示表。B. 同步会变化的数据与不会变化的数据，前面5个步骤一样是一样，在此不进行复述。 1. 手动全量同步参考同步不会变化的数据 2.创建最终目标表参考同步不会变化的数据 3. 根据数据的时间进行动态分区参考同步不会变化的数据 4. 简单验证数据参考同步不会变化的数据 5. 删除临时表参考同步不会变化的数据 6. 创建增量表执行ODPS SQL，创建一张增量表 --- 创建增量表 CREATE TABLE odps_test_inc LIKE odps_test; 7. 更改同步脚本更改后的同步脚本如下，请注意蓝色部分：因为太长不好截图，所以脚本模式这里的where的参数值使用： FROM_UNIXTIME ( created_at,'%Y%m%d')=${bdp.system.bizdate} OR FROM_UNIXTIME(updated_at, '%Y%m%d') = ${bdp.system.bizdate} 如果想效率更高一点可使用下方语句，当字段created_at有索引时可以命中： (created_at >= UNIX_TIMESTAMP('${bdp.system.bizdate}000000') AND created_at <= UNIX_TIMESTAMP('${bdp.system.bizdate}235959')) OR (updated_at >= UNIX_TIMESTAMP('${bdp.system.bizdate}000000') AND updated_at <= UNIX_TIMESTAMP('${bdp.system.bizdate}235959')) 8. 创建合并数据(ODPS_SQL)任务在数据开发->任务开发中创建一个节点任务，任务名称使用_merge结尾并输入以下内容： --- 合并数据并根据数据的创建时间更新到指定分区中 INSERT OVERWRITE TABLE odps_test PARTITION (pt) SELECT CASE WHEN b.id IS NOT NULL THEN b.uid ELSE a.uid END AS uid, CASE WHEN b.id IS NOT NULL THEN b.name ELSE a.name END AS name, CASE WHEN b.id IS NOT NULL THEN b.age ELSE a.age END AS age, CASE WHEN b.id IS NOT NULL THEN b.sex ELSE a.sex END AS sex, CASE WHEN b.id IS NOT NULL THEN b.created_at ELSE a.created_at END AS created_at, CASE WHEN b.id IS NOT NULL THEN b.updated_at ELSE a.updated_at END AS updated_at, CASE WHEN b.id IS NOT NULL THEN TO_CHAR(FROM_UNIXTIME(b.created_at), 'yyyymmdd') ELSE TO_CHAR(FROM_UNIXTIME(a.created_at), 'yyyymmdd') END AS pt FROM odps_test a FULL OUTER JOIN odps_test_inc b ON a.id = b.id ; --- 删除无用数据（如想观察每日数据变化不进行删除，上面一行的SQL需要加上条件： pt='${bdp.system.bizdate}') ALTER TABLE odps_test_inc DROP IF EXISTS PARTITION ( pt='${bdp.system.bizdate}'); 9. 设置调度依赖将合并数据的任务的上游任务设置为每日增量同步数据那个任务，并提交即可。注： odps_test是就是同步数据的脚本模式的任务名称。 10. 完成如果执行到这一步基本上就完成了，隔天就会在你在设定时间内运行，你也可以去运维中心中查看你的任务情况。下面附上”会变化的数据“数据同步的全部流程图：四、数据迁移 4.1 脚本模式下mongodb同步MaxCompute，时间字段值存的unix时间戳如何筛选? 这个问题我之前也有提交过工单，工单内容如下：我们在测试使用数据同步更新,使用query参数过滤数据，按文档中所描述，语法参照MongoDB查询语法https://help.aliyun.com/knowledge_detail/50354.html?spm=a2c1i.8282367.0.0.OPStre 因为我们mongodb表中表示数据"最后更新时间"的字段为updated_at,其存放的内容为unix时间戳。mongodb中将日期格式化成unix时间戳的方法是：ISODate("2012-10-15T21:26:17+0800").valueOf() / 1000 根据该写法，那在脚本模式中的写法将是："query": "{'updated_at':{'$gte':ISODate('${last_day}T00:00:00.424+0800').valueOf() / 1000}}", 但是我们运行后，程序报错无法运行了，报错信息为说是格式的问题，解析不了。报错信息如下:2017-06-23 20:46:32.647 [job-37098472] ERROR JobContainer - Exception when job run org.bson.json.JsonParseException: Invalid JSON input. Position: 62. Character: '.'. 当mongodb中字段为unix时间戳时，如果使用query参数进行过滤？我们花了时间查看文档依旧没有得到答案，所以前来请教一下。最终MaxCompute技术人员给的解决方案如下： dataworks 里面配置shell + datax，在shell 里面讲dataworks的调度时间参数yyyy-mm-dd格式转换为 unix时间戳格式，时间戳传递给datax，再由datax发给mongodb server 做数据过滤查询。在mongodb server 表里面添加时间列，时间列的值具体是你原来整数unix时间戳列生成的，这个只需要操作修改一次。 4.2 脚本模式下mongodb同步MaxCompute，同步速度如何优化？因为各种原因我们一开始是拿MongoDB进行的测试，为此专门找DBA挑一张较小的数据表( 约3GB左右)进行。按照文档中描述，使用脚本模式根据文档进行了配置，确实很挺简单的，所以进行任务测试，任务如期正常运行，但经过观察发现数据同步的速度逐步下降，最让人无法忍受的只有10KB/S，该表同步了20多个小时都没同步完，最后任务自行中止了。经过询问相关技术人员，可能由于Mongodb同步数据的底层实现还存在一些问题导致同步速度过慢导致的。为此只能选择替代方案，主要集中在“基于Kettle的MaxCompute插件”及“基于Tunnel SDK开发”，经过比对后考虑到开发速度及同步速度，最终选择暂时采用Kettle来进行mongodb的导入工作。如下为导入速度对比结果: 4.3 基于Kettle的MaxCompute插件同步mongodb遇到的问题以下为我的同事在使用中的经验总结，在此我一并提一下：MaxCompute的kettle插件的问题（个人只用过Aliyun MaxCompute Output，没有用过Aliyun MaxCompute Input，因此只对Output部分做评价）：没有提供更新的功能，只能插入，做增量更新的时候会很棘手。没有提供输出流，无法记录日志。不支持从上一步获取变量的功能，对于想要使用变量替换的操作很麻烦。 EndPoint、AccessId等配置不支持从配置文件读取，十分不灵活。文档太少，没有提供对MongoDB等数据库的帮助文档。 4.4 Mysql全库同步工具时间字段是unix时间戳没办法用。因为我们有大量的Mysql数据表需要导入，发现有个整库迁移的工具可以使用，刚发现时就怦然心动，然而经过实测后大失所望，因为该工具日期字段的值必须是yyyymmdd格式，像我们使用UNIX时间戳压根没法用。另外该功能不支持表字段指定，如果某些数据表字段过于敏感想不进行同步，也不能设置。 4.4 大数据开发套件中“数据表管理”如果表数量过多就不便于查找，且数据表分类只有一级。如果左侧能提供一个树形菜单将多级分类列出，点击就能查看对应分类下的表就方便多了。五、感谢接触一件新事务时遇到了困难，并且自己无法解决时，难免会感到彷徨无助，有种挫败感。万幸的是我在钉钉上找到了组织，非常感谢阿里云MaxCompute项目组的@数加·祎休(yī xiū)、@一斅、@彭敏、@李珍珍等工作人员的鼎力支持，在我们提交问题后，不厌其烦及时并且认真的回答解决我们的问题，祝你们工作顺利，事事顺心。

2017-07-23

Hive实践

准备 wget http://mirror.bit.edu.cn/apahce/hive/hive-0.11.0/hive-0.11.0-bin.tar.gz tar -zxvf hive-0.11.0-bin.tar.gz cd hive-0.11.0-bin sudo vi /etc/profile - export HIVE_HOME=/usr/hive-0.11.0-bin - export PATH =$PATH:$HIVE/bin hadoop fs -mkdir /tmp hadoop fs -mkdir /user/hive/warehouse hadoop fs -chmod g+w /tmp hadoop fs -chmod g+w /user/hive/warehouse cp conf/hive-log4j-properties.template conf/hive-log4j/properties local模式：set mapred.job.tracker=local; hadoop集群模式：set mapred.job.traker=host001:9001; hive 相关： create table demo (key int,value string) row format delimited fields terminated by '='stored as textfile; 加载数据到表demo: load data local inpath '/usr/hive-0.11.0-bin/data.txt'into table demo; 查询： select * from demo; select * from demo where key>=100 and key <=120; seclet *,count(*)as fre from demo group by value order by fre desc; 配置metastore 使用Mysql sudo apt-get install mysql-server mysql-client grant all privileges on *.* to 'root'@'%'with grant option; sudo vi /etc/mysql/my.cnf - 注释 bind-address =127.0.0.1 - mysql -uroot -psyc vi conf/hive-site.xml <?xml version = "1.0"?> <?xml-stylesheet type = "text/xsl" href ="configuration.xsl"?> <configuration> <property> <name>javax.jdo.option.ConnectionURL</name> <value>jdbc:mysql://host001:3306/hive?createDatabaseIfNotExis=true</value> </property> <name>javax.jdo.option.ConnectionDriverName</name> <value>com.mysql.jdbc.Driver</value> <property> </property> <name>javax.jdo.option.ConnectionUserName</name> <value>root</value> <property> <name>javax.jdo.option.ConnectionPassWord</name> <value>root</value> </property> </configuration>

2016-03-30

GitOps应用实践系列-Argo CD 实践篇

本篇我们将以一个示例项目为大家介绍 Argo CD 的实践。创建集群我们通过 KIND（Kubernetes in Docker）工具创建一个用于本地测试的 Kubernetes 集群。

2021-10-31

【实践篇】基于CAS的单点登录实践之路

想来单点登录有很多实现方案，不过最主流的还是基于CAS的方案，所以我也就分享一下我的CAS实践之路。什么是单点登录单点登录的英文名叫做：Single Sign On（简称SSO）。

2023-04-13

前端调试实践

其实只是抛砖引玉，还有很多东西是我们没有讲到的，我们可以通过下面的渠道进一步的了解、应用相关的知识和工具 7.1 Chrome DevTools 包含了Chrome 开发者工具相关功能的详细介绍还有相关的实践应用场景

2025-03-11

开源最佳实践

为什么要编写这一份《开源最佳实践》呢？首先，对于首次准备尝试参与开源的人来说，很容易有无从着手的感觉。完事开头难，希望这一份最佳实践可以帮忙更多有意参与开源的人们。

2021-06-08

Spark最佳实践

Spark最佳实践：在最后一章中，作者整理了大量Spark开发过程中的实践经验与应用技巧，从编码、监控、数据处理等几个大维度提供Spark应用程序开发的最佳实践思路参考。

2019-11-18

HBase最佳实践

本文致力于从架构原理、集群部署、性能优化与使用技巧等方面，阐述在如何基于HBase构建容纳大规模数据、支撑高并发、毫秒响应、稳定高效的OLTP实时系统。一、架构原理 1.1 基本架构从上层往下可以看到HBase架构中的角色分配为： Client Zookeeper HMaster RegionServer HDFS Client Client是执行查询、写入等对HBase表数据进行增删改查的使用方，可以是使用HBase Client API编写的程序，也可以是其他开发好的HBase客户端应用。 Zookeeper 同HDFS一样，HBase使用Zookeeper作为集群协调与管理系统。在HBase中其主要的功能与职责为：存储整个集群HMaster与RegionServer的运行状态实现HMaster的故障恢复与自动切换为Client提供元数据表的存储信息 HMaster、RegionServer启动之后将会在Zookeeper上注册并创建节点（/hbasae/master 与 /hbase/rs/*），同时 Zookeeper 通过Heartbeat的心跳机制来维护与监控节点状态，一旦节点丢失心跳，则认为该节点宕机或者下线，将清除该节点在Zookeeper中的注册信息。当Zookeeper中任一RegionServer节点状态发生变化时，HMaster都会收到通知，并作出相应处理，例如RegionServer宕机，HMaster重新分配Regions至其他RegionServer以保证集群整体可用性。当HMaster宕机时（Zookeeper监测到心跳超时），Zookeeper中的 /hbasae/master 节点将会消失，同时Zookeeper通知其他备用HMaster节点，重新创建 /hbasae/master 并转化为active master。协调过程示意图如下：除了作为集群中的协调者，Zookeeper还为Client提供了 hbase:meta 表的存储信息。客户端要访问HBase中的数据，只需要知道Zookeeper集群的连接信息，访问步骤如下：客户端将从Zookeeper（/hbase/meta-region-server）获得 hbase:meta 表存储在哪个RegionServer，缓存该位置信息查询该RegionServer上的 hbase:meta 表数据，查找要操作的 rowkey所在的Region存储在哪个RegionServer中，缓存该位置信息在具体的RegionServer上根据rowkey检索该Region数据可以看到，客户端操作数据过程并不需要HMaster的参与，通过Zookeeper间接访问RegionServer来操作数据。第一次请求将会产生3次RPC，之后使用相同的rowkey时客户端将直接使用缓存下来的位置信息，直接访问RegionServer，直至缓存失效（Region失效、迁移等原因）。通过Zookeeper的读写流程如下： hbase:meta 表 hbase:meta 表存储了集群中所有Region的位置信息。表结构如下： Rowkey格式：tableName,regionStartKey,regionId 第一个region的regionStartKey为空示例：ns1:testTable,,xxxxreigonid 只有一个列族info，包含三个列： regioninfo：RegionInfo的proto序列化格式，包含regionId,tableName,startKey,endKey,offline,split,replicaId等信息 server：RegionServer对应的server:port serverstartcode：RegionServer的启动时间戳简单总结Zookeeper在HBase集群中的作用如下：对于服务端，是实现集群协调与控制的重要依赖。对于客户端，是查询与操作数据必不可少的一部分。 HMaster HBase整体架构中HMaster的功能与职责如下：管理RegionServer，监听其状态，保证集群负载均衡且高可用。管理Region，如新Region的分配、RegionServer宕机时该节点Region的分配与迁移接收客户端的DDL操作，如创建与删除表、列簇等信息权限控制如我们前面所说的，HMaster 通过 Zookeeper 实现对集群中各个 RegionServer 的监控与管理，在 RegionServer 发生故障时可以发现节点宕机并转移 Region 至其他节点，以保证服务的可用性。但是HBase的故障转移并不是无感知的，相反故障转移过程中，可能会直接影响到线上请求的稳定性，造成段时间内的大量延迟。在分布式系统的 CAP定理中（Consistency一致性、Availability可用性、Partition tolerance分区容错性），分布式数据库基本特性都会实现P，但是不同的数据库对于A和C各有取舍。如HBase选择了C，而通过Zookeeper这种方式来辅助实现A（虽然会有一定缺陷），而Cassandra选择了A，通过其他辅助措施实现了C，各有优劣。对于HBase集群来说，HMaster是一个内部管理者，除了DDL操作并不对外（客户端）开放，因而HMaster的负载是比较低的。造成HMaster压力大的情况可能是集群中存在多个（两个或者三个以上）HMaster，备用的Master会定期与Active Master通信以获取最新的状态信息，以保证故障切换时自身的数据状态是最新的，因而Active Master可能会收到大量来自备用Master的数据请求。 RegionServer RegionServer在HBase集群中的功能与职责：根据HMaster的region分配请求，存放和管理Region 接受客户端的读写请求，检索与写入数据，产生大量IO 一个RegionServer中存储并管理者多个Region，是HBase集群中真正存储数据、接受读写请求的地方，是HBase架构中最核心、同时也是最复杂的部分。 RegionServer内部结构图如下： BlockCache BlockCache为RegionServer中的读缓存，一个RegionServer共用一个BlockCache。 RegionServer处理客户端读请求的过程： 1. 在BlockCache中查询是否命中缓存 2. 缓存未命中则定位到存储该数据的Region 3. 检索Region Memstore中是否有所需要的数据 4. Memstore中未查得，则检索Hfiles 5. 任一过程查询成功则将数据返回给客户端并缓存至BlockCache BlockCache有两种实现方式，有不同的应用场景，各有优劣： On-Heap的LRUBlockCache 优点：直接中Java堆内内存获取，响应速度快缺陷：容易受GC影响，响应延迟不稳定，特别是在堆内存巨大的情况下适用于：写多读少型、小内存等场景 Off-Heap的BucketCache 优点：无GC影响，延迟稳定缺陷：从堆外内存获取数据，性能略差于堆内内存适用于：读多写少型、大内存等场景我们将在「性能优化」一节中具体讨论如何判断应该使用哪种内存模式。 WAL 全称 Write Ahead Log ，是 RegionServer 中的预写日志。所有写入数据默认情况下都会先写入WAL中，以保证RegionServer宕机重启之后可以通过WAL来恢复数据，一个RegionServer中共用一个WAL。 RegionServer的写流程如下： 1. 将数据写入WAL中 2. 根据TableName、Rowkey和ColumnFamily将数据写入对应的Memstore中 3. Memstore通过特定算法将内存中的数据刷写成Storefile写入磁盘，并标记WAL sequence值 4. Storefile定期合小文件 WAL会通过日志滚动的操作定期对日志文件进行清理（已写入HFile中的数据可以清除），对应HDFS上的存储路径为 /hbase/WALs/${HRegionServer_Name} 。 Region 一个Table由一个或者多个Region组成，一个Region中可以看成是Table按行切分且有序的数据块，每个Region都有自身的StartKey、EndKey。一个Region由一个或者多个Store组成，每个Store存储该Table对应Region中一个列簇的数据，相同列簇的列存储在同一个Store中。同一个Table的Region会分布在集群中不同的RegionServer上以实现读写请求的负载均衡。故，一个RegionServer中将会存储来自不同Table的N多个Region。 Store、Region与Table的关系可以表述如下：多个Store（列簇）组成Region，多个Region（行数据块）组成完整的Table。其中，Store由Memstore（内存）、StoreFile（磁盘）两部分组成。在RegionServer中，Memstore可以看成指定Table、Region、Store的写缓存（正如BlockCache小节中所述，Memstore还承载了一些读缓存的功能），以RowKey、Column Family、Column、Timestamp进行排序。如下图所示：写请求到RegionServer之后并没有立刻写入磁盘中，而是先写入内存中的Memstore（内存中数据丢失问题可以通过回放WAL解决）以提升写入性能。 Region中的Memstore会根据特定算法将内存中的数据将会刷写到磁盘形成Storefile文件，因为数据在Memstore中为已排序，顺序写入磁盘性能高、速度快。在这种 Log-Structured Merge Tree架构模式下随机写入 HBase拥有相当高的性能。 Memstore刷磁盘形成的StoreFile 以HFile格式存储HBase的KV数据于HDFS之上。 HDFS HDFS为HBase提供底层存储系统，通过HDFS的高可用、高可靠等特性，保障了HBase的数据安全、容灾与备份。 1.2 写数据与 Memstore Flush 对于客户端来说，将请求发送到需要写入的RegionServer中，等待RegionServer写入WAL、Memstore之后即返回写入成功的ack信号。对于RegionServer来说，写入的数据还需要经过一系列的处理步骤。首先我们知道Memstore是在内存中的，将数据放在内存中可以得到优异的读写性能，但是同样也会带来麻烦：内存中的数据如何防止断电丢失将数据存储于内存中的代价是高昂的，空间总是有限的对于第一个问题，虽然可以通过WAL机制在重启的时候进行数据回放，但是对于第二个问题，则必须将内存中的数据持久化到磁盘中。在不同情况下，RegionServer通过不同级别的刷写策略对Memstore中的数据进行持久化，根据触发刷写动作的时机以及影响范围，可以分为不同的几个级别： Memstore级别：Region中任意一个MemStore达到了 hbase.hregion.memstore.flush.size 控制的上限（默认128MB），会触发Memstore的flush。 Region级别：Region中Memstore大小之和达到了 hbase.hregion.memstore.block.multiplier hbase.hregion.memstore.flush.size 控制的上限（默认 2 128M = 256M），会触发Memstore的flush。 RegionServer级别：Region Server中所有Region的Memstore大小总和达到了 hbase.regionserver.global.memstore.upperLimit ＊ hbase_heapsize 控制的上限（默认0.4，即RegionServer 40%的JVM内存），将会按Memstore由大到小进行flush，直至总体Memstore内存使用量低于 hbase.regionserver.global.memstore.lowerLimit ＊ hbase_heapsize 控制的下限（默认0.38，即RegionServer 38%的JVM内存）。 RegionServer中HLog数量达到上限：将会选取最早的 HLog对应的一个或多个Region进行flush（通过参数hbase.regionserver.maxlogs配置）。 HBase定期flush：确保Memstore不会长时间没有持久化，默认周期为1小时。为避免所有的MemStore在同一时间都进行flush导致的问题，定期的flush操作有20000左右的随机延时。手动执行flush：用户可以通过shell命令 flush ‘tablename’或者flush ‘region name’分别对一个表或者一个Region进行flush。 Memstore刷写时会阻塞线上的请求响应，由此可以看到，不同级别的刷写对线上的请求会造成不同程度影响的延迟：对于Memstore与Region级别的刷写，速度是比较快的，并不会对线上造成太大影响对于RegionServer级别的刷写，将会阻塞发送到该RegionServer上的所有请求，直至Memstore刷写完毕，会产生较大影响所以在Memstore的刷写方面，需要尽量避免出现RegionServer级别的刷写动作。数据在经过Memstore刷写到磁盘时，对应的会写入WAL sequence的相关信息，已经持久化到磁盘的数据就没有必要通过WAL记录的必要。 RegionServer会根据这个sequence值对WAL日志进行滚动清理，防止WAL日志数量太多，RegionServer启动时加载太多数据信息。同样，在Memstore的刷写策略中可以看到，为了防止WAL日志数量太多，达到指定阈值之后将会选择WAL记录中最早的一个或者多个Region进行刷写。 1.3 读数据与 Bloom Filter 经过前文的了解，我们现在可以知道HBase中一条数据完整的读取操作流程中，Client会和Zookeeper、RegionServer等发生多次交互请求。基于HBase的架构，一条数据可能存在RegionServer中的三个不同位置：对于刚读取过的数据，将会被缓存到BlockCache中对于刚写入的数据，其存在Memstore中对于之前已经从Memstore刷写到磁盘的，其存在于HFiles中 RegionServer接收到的一条数据查询请求，只需要从以上三个地方检索到数据即可，在HBase中的检索顺序依次是：BlockCache -> Memstore -> HFiles。其中，BlockCache、Memstore都是直接在内存中进行高性能的数据检索。而HFiles则是真正存储在HDFS上的数据：检索HFiles时会产生真实磁盘的IO操作 Memstore不停刷写的过程中，将会产生大量的HFile 如何在大量的HFile中快速找到所需要的数据呢？为了提高检索HFiles的性能，HBase支持使用 Bloom Fliter 对HFiles进行快读定位。 Bloom Filter（布隆过滤器）是一种数据结构，常用于大规模数据查询场景，其能够快速判断一个元素一定不在集合中，或者可能在集合中。 Bloom Filter由一个长度为m的位数组和 k个哈希函数组成。其工作原理如下：原始集合写入一个元素时，Bloom Filter同时将该元素经过k个哈希函数映射成k个数字，并以这些数字为下标，将位数组中对应下标的元素标记为1 当需要判断一个元素是否存在于原始集合中，只需要将该元素经过同样的 k个哈希函数得到k个数字取位数组中对应下标的元素，如果都为1，则表示元素可能存在如果存在其中一个元素为0，则该元素不可能存在于原始集合中因为哈希碰撞问题，不同的元素经过相同的哈希函数之后可能得到相同的值对于集合外的一个元素，如果经过 k个函数得到的k个数字，对应位数组中的元素都为1，可能是该元素存在于集合中也有可能是集合中的其他元素”碰巧“让这些下标对应的元素都标记为1，所以只能说其可能存在对于集合中的不同元素，如果经过k个函数得到的k个数字中，任意一个重复位数组中对应下标的元素会被覆盖，此时该下标的元素不能被删除（即归零）删除可能会导致其他多个元素在Bloom Filter表示不“存在” 由此可见，Bloom Filter中：位数组的长度m越大，误差率越小，而存储代价越大哈希函数的个数k越多，误差率越小，而性能越低 HBase中支持使用以下两种Bloom Filter： ROW：基于 Rowkey 创建的Bloom Filter ROWCOL：基于 Rowkey+Column 创建的Bloom Filter 两者的区别仅仅是：是否使用列信息作为Bloom Filter的条件。使用ROWCOL时，可以让指定列的查询更快，因为其通过Rowkey与列信息来过滤不存在数据的HFile，但是相应的，产生的Bloom Filter数据会更加庞大。而只通过Rowkey进行检索的查询，即使指定了ROWCOL也不会有其他效果，因为没有携带列信息。通过Bloom Filter（如果有的话）快速定位到当前的Rowkey数据存储于哪个HFile之后（或者不存在直接返回），通过HFile携带的 Data Block Index 等元数据信息可快速定位到具体的数据块起始位置，读取并返回（加载到缓存中）。这就是Bloom Filter在HBase检索数据的应用场景：高效判断key是否存在高效定位key所在的HFile 当然，如果没有指定创建Bloom Filter，RegionServer将会花费比较多的力气一个个检索HFile来判断数据是否存在。 1.4 HFile存储格式通过Bloom Filter快速定位到需要检索的数据所在的HFile之后的操作自然是从HFile中读出数据并返回。据我们所知，HFile是HDFS上的文件（或大或小都有可能），现在HBase面临的一个问题就是如何在HFile中快速检索获得指定数据？ HBase随机查询的高性能很大程度上取决于底层HFile的存储格式，所以这个问题可以转化为 HFile的存储格式该如何设计，才能满足HBase 快速检索的需求。生成一个HFile Memstore内存中的数据在刷写到磁盘时，将会进行以下操作：会先现在内存中创建空的Data Block数据块包含预留的Header空间。而后，将Memstore中的KVs一个个顺序写满该Block（一般默认大小为64KB）。如果指定了压缩或者加密算法，Block数据写满之后将会对整个数据区做相应的压缩或者加密处理。随后在预留的Header区写入该Block的元数据信息，如压缩前后大小、上一个block的offset、checksum 等。内存中的准备工作完成之后，通过HFile Writer输出流将数据写入到HDFS中，形成磁盘中的Data Block。为输出的Data Block生成一条索引数据，包括 {startkey、offset、size} 信息，该索引数据会被暂时记录在内存中的Block Index Chunk中。至此，已经完成了第一个Data Block的写入工作，Memstore中的 KVs 数据将会按照这个过程不断进行写入内存中的Data Block -> 输出到HDFS -> 生成索引数据保存到内存中的Block Index Chunk 流程。值得一提的是，如果启用了Bloom Filter，那么 Bloom Filter Data（位图数据）与 Bloom元数据（哈希函数与个数等）将会和 KVs 数据一样被处理：写入内存中的Block -> 输出到HDFS Bloom Data Block -> 生成索引数据保存到相对应的内存区域中。由此我们可以知道，HFile写入过程中，Data Block 和 Bloom Data Block 是交叉存在的。随着输出的Data Block越来越多，内存中的索引数据Block Index Chunk也会越来越大。达到一定大小之后（默认128KB）将会经过类似Data Block的输出流程写入到HDFS中，形成 Leaf Index Block （和Data Block一样，Leaf Index Block也有对应的Header区保留该Block的元数据信息）。同样的，也会生成一条该 Leaf Index Block 对应的索引记录，保存在内存中的 Root Block Index Chunk。 Root Index -> Leaf Data Block -> Data Block 的索引关系类似 B+树的结构。得益于多层索引，HBase可以在不读取整个文件的情况下查找数据。随着内存中最后一个 Data Block、Leaf Index Block 写入到HDFS，形成 HFile 的 Scanned Block Section。 Root Block Index Chunk 也会从内存中写入HDFS，形成 HFile 的 Load-On-Open Section 的一部分。至此，一个完整的HFile已经生成，如下图所示：检索HFile 生成HFile之后该如何使用呢？ HFile的索引数据（包括 Bloom Filter索引和数据索引信息）会在 Region Open 的时候被加载到读缓存中，之后数据检索经过以下过程：所有的读请求，如果读缓存和Memstore中不存在，那么将会检索HFile索引通过Bloom Filter索引（如果有设置Bloom Filter的话）检索Bloom Data以快速定位HFile是否存在所需数据定位到数据可能存在的HFile之后，读取该HFile的三层索引数据，检索数据是否存在存在则根据索引中的元数据找到具体的 Data Block 读入内存，取出所需的KV数据可以看到，在HFile的数据检索过程中，一次读请求只有真正确认数据存在且需要读取硬盘数据的时候才会执行硬盘查询操作。同时，得益于分层索引与分块存储，在Region Open加载索引数据的时候，再也不必和老版本（0.9甚至更早，HFile只有一层数据索引并且统一存储）一样加载所有索引数据到内存中导致启动缓慢甚至卡机等问题。 1.5 HFile Compaction Bloom Filter解决了如何在大量的HFile中快速定位数据所在的HFile文件，虽然有了Bloom Filter的帮助大大提升了检索效率，但是对于RegionServer来说要检索的HFile数量并没有减少。为了再次提高HFile的检索效率，同时避免大量小文件的产生造成性能低下，RegionServer会通过 Compaction机制对HFile进行合并操作。常见的Compaction触发方式有： Memstore Flush检测条件执行 RegionServer定期检查执行用户手动触发执行 Minor Compaction Minor Compaction 只执行简单的文件合并操作，选取较小的HFiles，将其中的数据顺序写入新的HFile后，替换老的HFiles。但是如何在众多HFiles中选择本次Minor Compaction要合并的文件却有不少讲究：首先排除掉文件大小大于 hbase.hstore.compaction.max.size 值的HFile 将HFiles按照文件年龄排序（older to younger），并从older file开始选择如果该文件大小小于 hbase.hstore.compaction.min 则加入Minor Compaction中如果该文件大小小于后续hbase.hstore.compaction.max 个HFile大小之和 * hbase.hstore.compaction.ratio，则将该文件加入Minor Compaction中扫描过程中，如果需要合并的HFile文件数达到 hbase.hstore.compaction.max（默认为10）则开始合并过程扫描结束后，如果需要合并的HFile的文件数大于 hbase.hstore.compaction.min（默认为3）则开始合并过程通过 hbase.offpeak.start.hour、hbase.offpeak.end.hour 设置高峰、非高峰时期，使 hbase.hstore.compaction.ratio的值在不同时期灵活变化（高峰值1.2、非高峰值5）可以看到，Minor Compaction不会合并过大的HFile，合并的HFile数量也有严格的限制，以避免产生太大的IO操作，Minor Compaction经常在Memstore Flush后触发，但不会对线上读写请求造成太大延迟影响。 Major Compaction 相对于Minor Compaction 只合并选择的一部分HFile合并、合并时只简单合并数据文件的特点，Major Compaction则将会把Store中的所有HFile合并成一个大文件，将会产生较大的IO操作。同时将会清理三类无意义数据：被删除的数据、TTL过期数据、版本号超过设定版本号的数据，Region Split过程中产生的Reference文件也会在此时被清理。 Major Compaction定期执行的条件由以下两个参数控制： hbase.hregion.majorcompaction：默认7天 hbase.hregion.majorcompaction.jitter：默认为0.2 集群中各个RegionServer将会在 hbase.hregion.majorcompaction +- hbase.hregion.majorcompaction * hbase.hregion.majorcompaction.jitter 的区间浮动进行Major Compaction，以避免过多RegionServer同时进行，造成较大影响。 Major Compaction 执行时机触发之后，简单来说如果当前Store中HFile的最早更新时间早于某个时间值，就会执行Major Compaction，该时间值为 hbase.hregion.majorcompaction * hbase.hregion.majorcompaction.jitter 。手动触发的情况下将会直接执行Compaction。 Compaction的优缺点 HBase通过Compaction机制使底层HFile文件数保持在一个稳定的范围，减少一次读请求产生的IO次数、文件Seek次数，确保HFiles文件检索效率，从而实现高效处理线上请求。如果没有Compaction机制，随着Memstore刷写的数据越来越多，HFile文件数量将会持续上涨，一次读请求生产的IO操作、Seek文件的次数将会越来越多，反馈到线上就是读请求延迟越来越大。然而，在Compaction执行过程中，不可避免的仍然会对线上造成影响。对于Major Compaction来说，合并过程将会占用大量带宽、IO资源，此时线上的读延迟将会增大。对于Minor Compaction来说，如果Memstore写入的数据量太多，刷写越来越频繁超出了HFile合并的速度即使不停地在合并，但是HFile文件仍然越来越多，读延迟也会越来越大 HBase通过 hbase.hstore.blockingStoreFiles（默认7）来控制Store中的HFile数量超过配置值时，将会堵塞Memstore Flush阻塞flush操作，阻塞超时时间为 hbase.hstore.blockingWaitTime 阻塞Memstore Flush操作将会使Memstore的内存占用率越来越高，可能导致完全无法写入简而言之，Compaction机制保证了HBase的读请求一直保持低延迟状态，但付出的代价是Compaction执行期间大量的读延迟毛刺和一定的写阻塞（写入量巨大的情况下）。 1.6 Region Split HBase通过 LSM-Tree架构提供了高性能的随机写，通过缓存、Bloom Filter、HFile与Compaction等机制提供了高性能的随机读。至此，HBase已经具备了作为一个高性能读写数据库的基本条件。如果HBase仅仅到此为止的话，那么其也只是个在架构上和传统数据库有所区别的数据库而已，作为一个高性能读写的分布式数据库来说，其拥有近乎可以无限扩展的特性。支持HBase进行自动扩展、负载均衡的是Region Split机制。 Split策略与触发条件在HBase中，提供了多种Split策略，不同的策略触发条件各不相同。如上图所示，不同版本中使用的默认策略在变化。 ConstantSizeRegionSplitPolicy 固定值策略，阈值默认大小 hbase.hregion.max.filesize 优点：简单实现缺陷：考虑片面，小表不切分、大表切分成很多Region，线上使用弊端多 IncreasingToUpperBoundRegionSplitPolicy 非固定阈值计算公式 min(R^2 * memstore.flush.size, region.split.size) R为Region所在的Table在当前RegionServer上Region的个数最大大小 hbase.hregion.max.filesize 优点：自动适应大小表，对于Region个数多的阈值大，Region个数少的阈值小缺陷：对于小表来说会产生很多小region SteppingSplitPolicy: 非固定阈值如果Region个数为1，则阈值为 memstore.flush.size * 2 否则为 region.split.size 优点：对大小表更加友好，小表不会一直产生小Region 缺点：控制力度比较粗可以看到，不同的切分策略其实只是在寻找切分Region时的阈值，不同的策略对阈值有不同的定义。切分点切分阈值确认完之后，首先要做的是寻找待切分Region的切分点。 HBase对Region的切分点定义如下： Region中最大的Store中，最大的HFile中心的block中，首个Rowkey 如果最大的HFile只有一个block，那么不切分（没有middle key）得到切分点之后，核心的切分流程分为 prepare - execute - rollback 三个阶段。 prepare阶段在内存中初始化两个子Region（HRegionInfo对象），准备进行切分操作。 execute阶段 execute阶段执行流程较为复杂，具体实施步骤为： RegionServer在Zookeeper上的 /hbase/region-in-transition 节点中标记该Region状态为SPLITTING。 HMaster监听到Zookeeper节点发生变化，在内存中修改此Region状态为RIT。在该Region的存储路径下创建临时文件夹 .split 父Region close，flush所有数据到磁盘中，停止所有写入请求。在父Region的 .split文件夹中生成两个子Region文件夹，并写入reference文件 reference是一个特殊的文件，体现在其文件名与文件内容上文件名组成：${父Region对应切分点所在的HFile文件}.${父Region} 文件内容：[splitkey]切分点rowkey，[top?]true/false，true为top上半部分，false为bottom下半部分根据reference文件名，可以快速找到对应的父Region、其中的HFile文件、HFile切分点，从而确认该子Region的数据范围数据范围确认完毕之后进行正常的数据检索流程（此时仍然检索父Region的数据）将子Region的目录拷贝到HBase根目录下，形成新的Region 父Regin通知修改 hbase:meta 表后下线，不再提供服务此时并没有删除父Region数据，仅在表中标记split列、offline列为true，并记录两个子region 两个子Region上线服务通知 hbase:meta 表标记两个子Region正式提供服务 rollback阶段如果execute阶段出现异常，则执行rollback操作，保证Region切分整个过程是具备事务性、原子性的，要么切分成功、要么回到未切分的状态。 region切分是一个复杂的过程，涉及到父region切分、子region生成、region下线与上线、zk状态修改、元数据状态修改、master内存状态修改等多个子步骤，回滚程序会根据当前进展到哪个子阶段清理对应的垃圾数据。为了实现事务性，HBase设计了使用状态机（SplitTransaction类）来保存切分过程中的每个子步骤状态。这样一来一旦出现异常，系统可以根据当前所处的状态决定是否回滚，以及如何回滚。但是目前实现中，中间状态是存储在内存中，因此一旦在切分过程中RegionServer宕机或者关闭，重启之后将无法恢复到切分前的状态。即Region切分处于中间状态的情况，也就是RIT。由于Region切分的子阶段很多，不同阶段解决RIT的处理方式也不一样，需要通过hbck工具进行具体查看并分析解决方案。好消息是HBase2.0之后提出了新的分布式事务框架Procedure V2，将会使用HLog存储事务中间状态，从而保证事务处理中宕机重启后可以进行回滚或者继续处理，从而减少RIT问题产生。父Region清理从以上过程中我们可以看到，Region的切分过程并不会父Region的数据到子Region中，只是在子Region中创建了reference文件，故Region切分过程是很快的。只有进行Major Compaction时才会真正（顺便）将数据切分到子Region中，将HFile中的kv顺序读出、写入新的HFile文件。 RegionServer将会定期检查 hbase:meta 表中的split和offline为true的Region，对应的子Region是否存在reference文件，如果不存在则删除父Region数据。负载均衡 Region切分完毕之后，RegionServer上将会存在更多的Region块，为了避免RegionServer热点，使请求负载均衡到集群各个节点上，HMaster将会把一个或者多个子Region移动到其他RegionServer上。移动过程中，如果当前RegionServer繁忙，HMaster将只会修改Region的元数据信息至其他节点，而Region数据仍然保留在当前节点中，直至下一次Major Compaction时进行数据移动。至此，我们已经揭开了HBase架构与原理的大部分神秘面纱，在后续做集群规划、性能优化与实际应用中，为什么这么调整以及为什么这么操作都将一一映射到HBase的实现原理上。如果你希望了解HBase的更多细节，可以参考《HBase权威指南》。二、集群部署经过冗长的理论初步了解过HBase架构与工作原理之后，搭建HBase集群是使用HBase的第一个步骤。需要注意的是，HBase集群一旦部署使用，再想对其作出调整需要付出惨痛代价（线上环境中），所以如何部署HBase集群是使用的第一个关键步骤。 2.1 集群物理架构硬件混合型+软件混合型集群硬件混合型指的是该集群机器配置参差不齐，混搭结构。软件混合型指的是该集群部署了一套类似CDH全家桶套餐。如以下的集群状况：集群规模：30 部署服务：HBase、Spark、Hive、Impala、Kafka、Zookeeper、Flume、HDFS、Yarn等硬件情况：内存、CPU、磁盘等参差不齐，有高配有低配，混搭结构这个集群不管是规模、还是服务部署方式相信都是很多都有公司的「标准」配置。那么这样的集群有什么问题呢？如果仅仅HBase是一个非「线上」的系统，或者充当一个历史冷数据存储的大数据库，这样的集群其实一点问题也没有，因为对其没有任何苛刻的性能要求。但是如果希望HBase作为一个线上能够承载海量并发、实时响应的系统，这个集群随着使用时间的增加很快就会崩溃。从硬件混合型来说，一直以来Hadoop都是以宣称能够用低廉、老旧的机器撑起一片天。这确实是Hadoop的一个大优势，然而前提是作为离线系统使用。离线系统的定义，即跑批的系统，如：Spark、Hive、MapReduce等，没有很强的时间要求，显著的吞吐量大，延迟高。因为没有实时性要求，几台拖拉机跑着也没有问题，只要最后能出结果并且结果正确就OK。那么在我们现在的场景中，对HBase的定义已经不是一个离线系统，而是一个实时系统。对于一个硬性要求很高的实时系统来说，如果其中几台老机器拖了后腿也会引起线上响应的延迟。统一高配硬件+软件混合型集群既然硬件拖后腿，那么硬件升级自然是水到渠成。现在我们有全新的高配硬件可以使用，参考如下：集群规模：30 部署服务：HBase、Spark、Hive、Impala、Kafka、Zookeeper、Flume、HDFS、Yarn等硬件情况：内存、CPU、磁盘统一高配置这样的集群可能还会存在什么问题呢？从软件混合型来说，离线任务最大的特点就是吞吐量特别高，瞬间读写的数据量可以把IO直接撑到10G/s，最主要的影响因素就是大型离线任务带动高IO将会影响HBase的响应性能。如果仅止步于此，那么线上的表现仅仅为短暂延迟，真正令人窒息的操作是，如果离线任务再把CPU撑爆，RegionServer节点可能会直接宕机，造成严重的生产影响。存在的另外一种情况是，离线任务大量读写磁盘、读写HDFS，导致HBase IO连接异常也会造成RegionServer异常（HBase日志反应HDFS connection timeout，HDFS日志反应IO Exception），造成线上故障。根据观测，集群磁盘IO到4G以上、集群网络IO 8G以上、HDFS IO 5G以上任意符合一个条件，线上将会有延迟反应。因为离线任务运行太过强势导致RegionServer宕机无法解决，那么能采取的策略只能是重新调整离线任务的执行使用资源、执行顺序等，限制离线计算能力来满足线上的需求。同时还要限制集群的CPU的使用率，可能出现某台机器CPU打满后整个机器假死致服务异常，造成线上故障。软、硬件独立的HBase集群简而言之，无论是硬件混合型还是软件混合型集群，其可能因为各种原因带来的延迟影响，对于一个高性能要求的HBase来说，都是无法忍受的。所以在集群规划初始就应该考虑到种种情况，最好使用独立的集群部署HBase。参考如下一组集群规模配置：集群规模：15+5（RS+ZK）部署服务：HBase、HDFS（另5台虚拟Zookeeper）硬件情况：除虚拟机外，物理机统一高配置虽然从可用节点上上来看比之前的参考配置少了一半，但是从集群部署模式上看，最大程度保证HBase的稳定性，从根本上分离了软硬件对HBase所带来的影响，将会拥有比之前两组集群配置更稳定的响应和更高的性能。其他硬件推荐网卡：网卡是容易产生瓶颈的地方，有条件建议使用双万兆网卡磁盘：没有特殊要求，空间越大越好，转速越高越好内存：不需要大容量内存，建议32-128G（详见下文） CPU：CPU核数越多越好，HBase本身压缩数据、合并HFile等都需要CPU资源。电源：建议双电源冗余另外值得注意的是，Zookeeper节点建议设置5个节点，5个节点能保证Leader快速选举，并且最多可以允许2个节点宕机的情况下正常使用。硬件上可以选择使用虚拟机，因为zk节点本身消耗资源并不大，不需要高配机器。但是5个虚拟节点不能在一个物理机上，防止物理机宕机影响所有zk节点。 2.2 安装与部署以CDH集群为例安装HBase。使用自动化脚本工具进行安装操作： # 获取安装脚本，上传相关安装软件包至服务器（JDK、MySQL、CM、CDH等） yum install -y git git clone https://github.com/chubbyjiang/cdh-deploy-robot.git cd cdh-deploy-robot # 编辑节点主机名 vi hosts # 修改安装配置项 vi deploy-robot.cnf # 执行 sh deploy-robot.sh install_all 安装脚本将会执行配置SSH免密登录、安装软件、操作系统优化、Java等开发环境初始化、MySQL安装、CM服务安装、操作系统性能测试等过程。脚本操作说明见：CDH集群自动化部署工具。等待cloudera-scm-server进程起来后，在浏览器输入 ip:7180 进入CM管理界面部署HDFS、HBase组件即可。三、性能优化 HBase集群部署完毕运行起来之后，看起来一切顺利，但是所有东西都处于「初始状态」中。我们需要根据软硬件环境，针对性地对HBase进行调优设置，以确保其能够以最完美的状态运行在当前集群环境中，尽可能发挥硬件的优势。为了方便后续配置项计算说明，假设我们可用的集群硬件状况如下：总内存：256G 总硬盘：1.8T * 12 = 21.6T 可分配内存：256 * 0.75 = 192G HBase可用内存空间：192 * 0.8 = 153G（20%留给HDFS等其他进程）可用硬盘空间：21.6T * 0.85 = 18.36T 3.1 Region规划对于Region的大小，HBase官方文档推荐单个在10G-30G之间，单台RegionServer的数量控制在20-300之间（当然，这仅仅是参考值）。 Region过大过小都会有不良影响：过大的Region 优点：迁移速度快、减少总RPC请求缺点：compaction的时候资源消耗非常大、可能会有数据分散不均衡的问题过小的Region 优点：集群负载平衡、HFile比较少compaction影响小缺点：迁移或者balance效率低、频繁flush导致频繁的compaction、维护开销大规划Region的大小与数量时可以参考以下算法： 0. 计算HBase可用磁盘空间（单台RegionServer） 1. 设置region最大与最小阈值，region的大小在此区间选择，如10-30G 2. 设置最佳region数（这是一个经验值），如单台RegionServer 200个 3. 从region最小值开始，计算 HBase可用磁盘空间 / (region_size * hdfs副本数) = region个数 4. 得到的region个数如果 > 200，则增大region_size（step可设置为5G），继续计算直至找到region个数最接近200的region_size大小 5. region大小建议不小于10G 当前可用磁盘空间为18T，选择的region大小范围为10-30G，最佳region个数为300。那么最接近最佳Region个数300的 region_size 值为30G。得到以下配置项： hbase.hregion.max.filesize=30G 单节点最多可存储的Region个数约为300 3.2 内存规划我们知道RegionServer中的BlockCache有两种实现方式： LRUBlockCache：On-Heap BucketCache：Off-Heap 这两种模式的详细说明可以参考 CDH官方文档。为HBase选择合适的内存模式以及根据内存模式计算相关配置项是调优中的重要步骤。首先我们可以根据可用内存大小来判断使用哪种内存模式。先看超小内存（假设8G以下）和超大内存（假设128G以上）两种极端情况：对于超小内存来说，即使可以使用BucketCache来利用堆外内存，但是使用堆外内存的主要目的是避免GC时不稳定的影响，堆外内存的效率是要比堆内内存低的。由于内存总体较小，即使读写缓存都在堆内内存中，GC时也不会造成太大影响，所以可以直接选择LRUBlockCache。对于超大内存来说，在超大内存上使用LRUBlockCache将会出现我们所担忧的情况：GC时对线上造成很不稳定的延迟影响。这种场景下，应该尽量利用堆外内存作为读缓存，减小堆内内存的压力，所以可以直接选择BucketCache。在两边的极端情况下，我们可以根据内存大小选择合适的内存模式，那么如果内存大小在合理、正常的范围内该如何选择呢？此时我们应该主要关注业务应用的类型。当业务主要为写多读少型应用时，写缓存利用率高，应该使用LRUBlockCache尽量提高堆内写缓存的使用率。当业务主要为写少读多型应用时，读缓存利用率高（通常也意味着需要稳定的低延迟响应），应该使用BucketCache尽量提高堆外读缓存的使用率。对于不明确或者多种类型混合的业务应用，建议使用BucketCache，保证读请求的稳定性同时，堆内写缓存效率并不会很低。当前HBase可使用的内存高达153G，故将选择BucketCache的内存模型来配置HBase，该模式下能够最大化利用内存，减少GC影响，对线上的实时服务较为有利。得到配置项： hbase.bucketcache.ioengine=offheap: 使用堆外缓存确认使用的内存模式之后，接下来将通过计算确认 JavaHeap、对外读缓存、堆内写缓存、LRU元数据等内存空间具体的大小。内存与磁盘比讨论具体配置之前，我们从 HBase集群规划引入一个Disk / JavaHeap Ratio的概念来帮助我们设置内存相关的参数。理论上我们假设最优情况下硬盘维度下的Region个数和 JavaHeap维度下的Region个数相等。相应的计算公式为：硬盘容量维度下Region个数： DiskSize / (RegionSize ＊ ReplicationFactor) JavaHeap维度下Region个数： JavaHeap * HeapFractionForMemstore / (MemstoreSize / 2 ) 其中： RegionSize：Region大小，配置项：hbase.hregion.max.filesize ReplicationFactor：HDFS的副本数，配置项：dfs.replication HeapFractionForMemstore：JavaHeap写缓存大小，即RegionServer内存中Memstore的总大小，配置项：hbase.regionserver.global.memstore.lowerLimit MemstoreSize：Memstore刷写大小，配置项：hbase.hregion.memstore.flush.size 现在我们已知条件硬盘维度和JavaHeap维度相等，求 1 bytes的JavaHeap大小需要搭配多大的硬盘大小。已知： DiskSize / (RegionSize ＊ ReplicationFactor) = JavaHeap * HeapFractionForMemstore / (MemstoreSize / 2 ) 求： DiskSize / JavaHeap 进行简单的交换运算可得： DiskSize / JavaHeap = RegionSize / MemstoreSize * ReplicationFactor * HeapFractionForMemstore * 2 以HBase的默认配置为例： RegionSize: 10G MemstoreSize: 128M ReplicationFactor: 3 HeapFractionForMemstore: 0.4 计算： 10G / 128M * 3 * 0.4 * 2 = 192 即理想状态下 RegionServer上 1 bytes的Java内存大小需要搭配192bytes的硬盘大小最合适。套用到当前集群中，HBase可用内存为152G，在LRUBlockCache模式下，对应的硬盘空间需要为153G * 192 = 29T，这显然是比较不合理的。在BucketCache模式下，当前 JavaHeap、HeapFractionForMemstore 等值还未确定，我们会根据这个计算关系和已知条件对可用内存进行规划和调整，以满足合理的内存/磁盘比。已知条件：内存模式：BucketCache 可用内存大小：153G 可用硬盘大小：18T Region大小：30G ReplicationFactor：3 未知变量： JavaHeap MemstoreSize HeapFractionForMemstore 内存布局在计算位置变量的具体值之前，我们有必要了解一下当前使用的内存模式中对应的内存布局。 BucketCache模式下，RegionServer的内存划分如下图：简化版：写缓存从架构原理中我们知道，Memstore有4种级别的Flush，需要我们关注的是 Memstore、Region和RegionServer级别的刷写。其中Memstore和Region级别的刷写并不会对线上造成太大影响，但是需要控制其阈值和刷写频次来进一步提高性能。而RegionServer级别的刷写将会阻塞请求直至刷写完成，对线上影响巨大，需要尽量避免。得到以下配置项： hbase.hregion.memstore.flush.size=256M: 控制的Memstore大小默认值为128M，太过频繁的刷写会导致IO繁忙，刷新队列阻塞等。设置太高也有坏处，可能会较为频繁的触发RegionServer级别的Flush，这里设置为256M。 hbase.hregion.memstore.block.multiplier=3: 控制的Region flush上限默认值为2，意味着一个Region中最大同时存储的Memstore大小为2 * MemstoreSize ，如果一个表的列族过多将频繁触发，该值视情况调整。现在我们设置两个经验值变量： RegionServer总内存中JavaHeap的占比=0.35 JavaHeap最大大小=56G：超出此值表示GC有风险计算得JavaHeap的大小为 153 * 0.35 = 53.55 ，没有超出预期的最大JavaHeap。如果超过最大期望值，则使用最大期望值代替，得JavaHeap大小为53G。现在JavaHeap、MemstoreSize已知，可以得到唯一的位置变量 HeapFractionForMemstore 的值为 0.48 。得到以下配置项： RegionServer JavaHeap堆栈大小: 53G hbase.regionserver.global.memstore.upperLimit=0.58: 整个RS中Memstore最大比例，比lower大5-15% hbase.regionserver.global.memstore.lowerLimit=0.48: 整个RS中Memstore最小比例写缓存大小为 53 * 0.48 = 25.44G 读缓存配置当前内存信息如下： A 总可用内存：153G J JavaHeap大小：53G W 写缓存大小：25.44G R1 LRU缓存大小：？ R2 BucketCache堆外缓存大小：153 - 53 = 100G 因为读缓存由堆内的LRU元数据与堆外的数据缓存组成，两部分占比一般为 1:9（经验值）。而对于总体的堆内内存，存在以下限制，如果超出此限制则应该调低比例： LRUBlockCache + MemStore < 80% * JVM_HEAP 即 LRUBlockCache + 25.44 < 53 * 0.8 可得R1的最大值为16.96G 总读缓存：R = R1 + R2 R1:R2 = 1:9 R1 = 11G < 16G R = 111G 配置堆外缓存涉及到的相关参数如下： hbase.bucketcache.size=111 * 1024M: 堆外缓存大小，单位为M hbase.bucketcache.percentage.in.combinedcache=0.9: 堆外读缓存所占比例，剩余为堆内元数据缓存大小 hfile.block.cache.size=0.15: 校验项，+upperLimit需要小于0.8 现在，我们再来计算 Disk / JavaHeap Ratio 的值，检查JavaHeap内存与磁盘的大小是否合理： RegionSize / MemstoreSize * ReplicationFactor * HeapFractionForMemstore * 2 30 * 1024 / 256 * 3 * 0.48 * 2 = 345.6 53G * 345.6 = 18T <= 18T 至此，已得到HBase中内存相关的重要参数： RegionServer JavaHeap堆栈大小: 53G hbase.hregion.max.filesize=30G hbase.bucketcache.ioengine=offheap hbase.hregion.memstore.flush.size=256M hbase.hregion.memstore.block.multiplier=3 hbase.regionserver.global.memstore.upperLimit=0.58 hbase.regionserver.global.memstore.lowerLimit=0.48 hbase.bucketcache.size=111 * 1024M hbase.bucketcache.percentage.in.combinedcache=0.9 hfile.block.cache.size=0.15 3.3 合并与切分 HFile合并 Compaction过程中，比较常见的优化措施是： Major Compaction 停止自动执行增大其处理线程数 Minor Compaction 增加Memstore Flush大小增加Region中最大同时存储的Memstore数量配置项如下： # 关闭major compaction，定时在业务低谷执行，每周一次 hbase.hregion.majorcompaction=0 # 提高compaction的处理阈值 hbase.hstore.compactionThreshold=6 # 提高major compaction处理线程数 hbase.regionserver.thread.compaction.large=5 # 提高阻塞memstore flush的hfile文件数阈值 hbase.hstore.blockingStoreFiles=100 hbase.hregion.memstore.flush.size=256M hbase.hregion.memstore.block.multiplier=3: hbase.hstore.compaction.minhbase.hstore.compaction.max--> Major Compaction 脚本关闭自动compaction之后手动执行脚本的代码示例： #!/bin/bash if [ $# -lt 1 ] then echo "Usage: <table key>" exit 1 fi TMP_FILE=tmp_tables TABLES_FILE=tables.txt key=$1 echo "list" | hbase shell > $TMP_FILE sleep 2 sed '1,6d' $TMP_FILE | tac | sed '1,2d' | tac | grep $key > $TABLES_FILE sleep 2 for table in $(cat $TABLES_FILE); do date=`date "+%Y%m%d %H:%M:%S"` echo "major_compact '$table'" | hbase shell echo "'$date' major_compact '$table'" >> /tmp/hbase-major-compact.log sleep 5 done rm -rf $TMP_FILE rm -rf $TABLES_FILE echo "" >> /tmp/hbase-major-compact.log Region切分在架构原理中我们知道，Region多有种切分策略，在Region切分时将会有短暂时间内的Region下线无服务，Region切分完成之后的Major Compaction中，将会移动父Region的数据到子Region中，HMaster为了集群整体的负载均衡可能会将子Region分配到其他RegionServer节点。从以上描述中可以看到，Region的切分行为其实是会对线上的服务请求带来一定影响的。 Region切分设置中，使用默认配置一般不会有太大问题，但是有没有保证数据表负载均衡的情况下，Region不进行切分行为？有一种解决方案是使用预分区 + 固定值切分策略可以一定程度上通过预估数据表数量以及Region个数，从而在一段时间内抑制Region不产生切分。假设我们可以合理的预判到一个表的当前总数据量为150G，每日增量为1G，当前Region大小为30G。那么我们建表的时候至少要设定 (150 + 1 * 360) / 30 = 17 个分区，如此一来一年内（360天）该表的数据增长都会落到17个Region中而不再切分。当然对于一个不断增长的表，除非时间段设置的非常长，否则总有发生切分的一天。如果无限制的延长时间段则会在一开始就产生大量的空Region，这对HBase是极其不友好的，所以时间段是一个需要合理控制的阈值。在hbase-site.xml中配置Region切分策略为ConstantSizeRegionSplitPolicy： hbase.regionserver.region.split.policy=org.apache.hadoop.hbase.regionserver.ConstantSizeRegionSplitPolicy 3.4 响应优化 HBase服务端高并发情况下，如果HBase服务端处理线程数不够，应用层将会收到HBase服务端抛出的无法创建新线程的异常从而导致应用层线程阻塞。可以释放调整HBase服务端配置以提升处理性能： # Master处理客户端请求最大线程数 hbase.master.handler.count=256 # RS处理客户端请求最大线程数，如果该值设置过大则会占用过多的内存，导致频繁的GC，或者出现OutOfMemory hbase.regionserver.handler.count=256 # 客户端缓存大小，默认为2M hbase.client.write.buffer=8M # scan缓存一次获取数据的条数，太大也会产生OOM hbase.client.scanner.caching=100 另外，以下两项中，默认设置下超时太久、重试次数太多，一旦应用层连接不上HBse服务端将会进行近乎无限的重试，长连接无法释放，新请求不断进来，从而导致线程堆积应用假死等，影响比较严重，可以适当减少： hbase.client.retries.number=3 hbase.rpc.timeout=10000 HDFS 适当增加处理线程等设置： dfs.datanode.handler.count=64 dfs.datanode.max.transfer.threads=12288 dfs.namenode.handler.count=256 dfs.namenode.service.handler.count=256 同时，对于HDFS的存储设置也可以做以下优化： # 可以配置多个，拥有多个元数据备份 dfs.name.dir # 配置多个磁盘与路径，提高并行读写能力 dfs.data.dir # dn同时处理文件的上限，默认为256，可以提高到8192 dfs.datanode.max.xcievers 应用层（客户端）之前我们说到，HBase为了保证CP，在A的实现上做了一定的妥协，导致HBase出现故障并转移的过程中会有较大的影响。对于应用服务层来说，保证服务的稳定性是最重要的，为了避免HBase可能产生的问题，应用层应该采用读写分离的模式来最大程度保证自身稳定性。应用层读写分离可靠的应用层应使用读写分离的模式提高响应效率与可用性：读写应用应该分别属于不同的服务实例，避免牵一发而动全身对于写入服务，数据异步写入redis或者kafka队列，由下游消费者同步至HBase，响应性能十分优异需要处理数据写入失败的事务处理与重写机制对于读取服务，如果一个RS挂了，一次读请求经过若干重试和超时可能会持续几十秒甚至更久，由于和写入服务分离可以做到互不影响最好使用缓存层来环节RS宕机问题，对于至关重要的数据先查缓存再查HBase（见下文）在应用层的代码中，同样有需要注意的小TIPS：如果在Spring中将HBaseAdmin配置为Bean加载，则需配置为懒加载，避免在启动时链接HMaster失败导致启动失败，从而无法进行一些降级操作。 scanner使用后及时关闭，避免浪费客户端和服务器的内存查询时指定列簇或者指定要查询的列限定扫描范围 Put请求可以关闭WAL，但是优化不大最后，可以适当调整一下连接池设置： # 配置文件加载为全局共享，可提升tps setInt(“hbase.hconnection.threads.max”, 512); setInt(“hbase.hconnection.threads.core”, 64); 3.5 使用缓存层即使我们经过大量的准备、调优与设置，在真实使用场景中，随着HBase中承载的数据量越来越大、请求越来越多、并发越来越大，HBase不可避免的会有一些「毛刺」问题。如果你现在已经通过HBase解决了大部分的线上数据存储与访问问题，但是有一小部分的数据需要提供最快速的响应、最低的延迟，由于HBase承载的东西太多，总是有延迟比较高的响应，此时需要怎么解决？其实，对所有数据库软件来说都会存在这样的场景。于是，类似关系型数据库中的数据库拆分等策略也是可以应用到HBase上的。或者是将最关键、最热点的数据使用独立的HBase集群来处理，或者是使用诸如 Redis等更高性能的缓存软件，其核心思想就是将最关键的业务数据独立存储以提供最优质的服务，这个服务统称为缓存层。 3.6 其他配置 hbase-env.sh 的 HBase 客户端环境高级配置代码段配置了G1垃圾回收器和其他相关属性： -XX:+UseG1GC -XX:InitiatingHeapOccupancyPercent=65 -XX:-ResizePLAB -XX:MaxGCPauseMillis=90 -XX:+UnlockDiagnosticVMOptions -XX:+G1SummarizeConcMark -XX:+ParallelRefProcEnabled -XX:G1HeapRegionSize=32m -XX:G1HeapWastePercent=20 -XX:ConcGCThreads=4 -XX:ParallelGCThreads=16 -XX:MaxTenuringThreshold=1 -XX:G1MixedGCCountTarget=64 -XX:+UnlockExperimentalVMOptions -XX:G1NewSizePercent=2 -XX:G1OldCSetRegionThresholdPercent=5 hbase-site.xml 的 RegionServer 高级配置代码段（安全阀）手动split region配置 <property><name>hbase.regionserver.wal.codec</name><value>org.apache.hadoop.hbase.regionserver.wal.IndexedWALEditCodec</value></property><property><name>hbase.region.server.rpc.scheduler.factory.class</name><value>org.apache.hadoop.hbase.ipc.PhoenixRpcSchedulerFactory</value><description>Factory to create the Phoenix RPC Scheduler that uses separate queues for index and metadata updates</description></property><property><name>hbase.rpc.controllerfactory.class</name><value>org.apache.hadoop.hbase.ipc.controller.ServerRpcControllerFactory</value><description>Factory to create the Phoenix RPC Scheduler that uses separate queues for index and metadata updates</description></property><property><name>hbase.regionserver.thread.compaction.large</name><value>5</value></property><property><name>hbase.regionserver.region.split.policy</name><value>org.apache.hadoop.hbase.regionserver.ConstantSizeRegionSplitPolicy</value></property> 四、使用技巧 4.1 建表规约 Rowkey规范如无特殊情况，长度应控制在64字节内。充分分析业务需求后确认需要查询的维度字段。 get请求，则rowkey散列处理。 scan请求，rowkey前缀维度散列后，后续维度依照查询顺序或者权重拼接（视具体情况决定是否散列处理）。各个字段都保持相同长度以支持左对齐的部分键扫描。 scan形式的数据表中，需要提前统计单个scan可扫描出的最大数量。列簇规范如无特殊情况，一个表中只有一个列簇，统一使用info命名。如果需要1以上的列簇，则原则上一次请求的数据不可跨列簇存储，多不超过3个列簇。示例：NAME =>'info' 压缩统一使用SNAPPY压缩。示例：COMPRESSION => 'SNAPPY' 版本默认版本数为3，前期存储空间紧张的情况下设置为1。示例：VERSIONS => 1 布隆过滤器视情情况使用，主要针对get查询提高性能。 kv示例：BLOOMFILTER => 'ROW'，根据rowkey中的信息生成布隆过滤器数据。 kv+col示例：BLOOMFILTER => 'ROWCOL'，根据rowkey+列信息生成布隆过滤器，针对get+指定列名的查询，产生的过滤器文件会比ROW大。预分区预分区需要通过评估整体表数据量来确认，当前hbase集群region块大小为30G。历史大增量小的数据：给定的预分区数足够支撑该表永远（或者相当长的时间内）不split，即更新的所有数据将进入已存在的region中，以减少split与compaction造成的影响。历史小增量大的数据：预分区个数需满足历史数据等分存储，并支撑未来一段时间内（一个月以上）的增量数据。预分区区间计算：属性相同的表中随机取出部分样本数据（rowkey维度字段）。将样本转换成rowkey之后排序，并以样本个数/预分区个数为步长，取预分区个数个rowkey组成预分区区间。预分区代码示例： /** * hbase region预分区工具 * * @param filePath 样本文件路径 * @param numOfSPlits 预分区个数 **/ def rowkeySplitedArr(filePath: String, numOfSPlits: Int) = { val file = Source.fromFile(filePath).getLines() val res = file.map { line => val arr = line.split("_") val card = arr(0) val name = arr(1) MathUtil.MD5Encrypt32(card) + MathUtil.MD5Encrypt32(card) }.toList.sorted val count = res.length / numOfSPlits var str = "" for (i <- 0 until numOfSPlits) { str += s"\'${res(i * count)}\'," } println(str.substring(0, str.length - 1)) } 4.2 客户端使用服务端配置完成之后，如何更好的使用HBase集群也需要花点心思测试与调整。以Spark作为HBase读写客户端为例。查询场景批量查询 Spark有对应的API可以批量读取HBase数据，但是使用过程比较繁琐，这里安利一个小组件Spark DB Connector，批量读取HBase的代码可以这么简单： val rdd = sc.fromHBase[(String, String, String)]("mytable") .select("col1", "col2") .inColumnFamily("columnFamily") .withStartRow("startRow") .withEndRow("endRow") done！实时查询以流式计算为例，Spark Streaming中，我们要实时查询HBase只能通过HBase Client API（没有队友提供服务的情况下）。那么HBase Connection每条数据创建一次肯定是不允许的，效率太低，对服务压力比较大，并且ZK的连接数会暴增影响服务。比较可行的方案是每个批次创建一个链接（类似foreachPartiton中每个分区创建一个链接，分区中数据共享链接）。但是这种方案也会造成部分连接浪费、效率低下等。如果可以做到一个Streaming中所有批次、所有数据始终复用一个连接池是最理想的状态。 Spark中提供了Broadcast这个重要工具可以帮我们实现这个想法，只要将创建的HBase Connection广播出去所有节点就都能复用，但是真实运行代码时你会发现HBase Connection是不可序列化的对象，无法广播。。。其实利用scala的lazy关键字可以绕个弯子来实现： //实例化该对象，并广播使用 class HBaseSink(zhHost: String, confFile: String) extends Serializable { //延迟加载特性 lazy val connection = { val hbaseConf = HBaseConfiguration.create() hbaseConf.set(HConstants.ZOOKEEPER_QUORUM, zhHost) hbaseConf.addResource(confFile) val conn = ConnectionFactory.createConnection(hbaseConf) sys.addShutdownHook { conn.close() } conn } } 在Driver程序中实例化该对象并广播，在各个节点中取广播变量的value进行使用。广播变量只在具体调用value的时候才会去创建对象并copy到各个节点，而这个时候被序列化的对象其实是外层的HBaseSink，当在各个节点上具体调用connection进行操作的时候，Connection才会被真正创建（在当前节点上），从而绕过了HBase Connection无法序列化的情况（同理也可以推导RedisSink、MySQLSink等）。这样一来，一个Streaming Job将会使用同一个数据库连接池，在Structured Streaming中的foreachWrite也可以直接应用。写入场景批量写入同理安利组件 rdd.toHBase("mytable") .insert("col1", "col2") .inColumnFamily("columnFamily") .save() 这里边其实对HBase Client的Put接口包装了一层，但是当线上有大量实时请求，同时线下又有大量数据需要更新时，直接这么写会对线上的服务造成冲击，具体表现可能为持续一段时间的短暂延迟，严重的甚至可能会把RS节点整挂。大量写入的数据带来具体大GC开销，整个RS的活动都被阻塞了，当ZK来监测心跳时发现无响应就将该节点列入宕机名单，而GC完成后RS发现自己“被死亡”了，那么就干脆自杀，这就是HBase的“朱丽叶死亡”。这种场景下，使用bulkload是最安全、快速的，唯一的缺点是带来的IO比较高。大批量写入更新的操作，建议使用bulkload工具来实现。实时写入理同实时查询，可以使用创建的Connection做任何操作。结束语我们从HBase的架构原理出发，接触了HBase大部分的核心知识点。理论基础决定上层建筑，有了对HBase的总体认知，在后续的集群部署、性能优化以及实际应用中都能够比较游刃有余。知其然而之所以然，保持对技术原理的探索，不仅能学习到其中许多令人惊叹的设计与操作，最重要的是能够真正在业务应用中充分发挥其应有的性能。 ### 4.3 异常与故障排查hadoop fs -rmr /hbase/WALsrmr /hbase hbase.regionserver.executor.openregion.threads 100 yarn进程问题buffer/cache太多手动启动执行（补充环境信息）僵尸进程发现并处理-->

2019-11-18

Kubernetes监控实践

四、K8s监控实践中的挑战 2014年上线以来，K8s一直在变革容器技术，已经成为快速批量启动应用的关键工具。与此同时，挑战也随之而来，容器编排极其复杂。

2019-09-19

IAM最佳实践

后面会有专门的文章来讲如何实践联合身份。统一账单云账户接收包括所有用户的资源操作所发生费用的统一账单。

2019-03-18

Kotlin实践记录

view.setOnClickListener { itemClickListener(pictures) view.context.longToast(pictures.url) } } } } 实践的代码见我的

2018-02-22

Docker实践（一）

Docker实践什么是LXC？ LXC为Linux Container的简写。

2017-11-11

Docker实践（二）

registrylatestc9bd19d022f6 5 weeksago 33.27 MB [root@localhost ~]# mkdir /opt/data/registry –p #运行创建私有仓库 registry的介绍提到的最佳实践建议将

2017-11-11

Docker最佳实践

本文讲的是Docker最佳实践，【编者的话】本文是Docker使用过程中的一些最佳实践。

2017-09-13

MaxCompute实践分析

概述本文主要是记录如何通过数加MaxCompute和大数据开发套件两个产品实现简单的电影数据分析。目标希望了解每天市场上最火爆的电影是哪些，表现如何。通过对影视及票房数据集（dwd_product_movie_basic_info电影基本信息和 ods_product_movie_box票房基本信息）统计分析出每天每个国家/地区累计票房最高的10个电影并且展现这些电影的导演、主演和累计票房。数据说明直接使用了数加平台公开数据集上的影视及票房数据集。该数据集从2016 年12 月13 日开始，每日早10 点前进行全量更新。两个表的具体信息如下：电影基本信息 dwd_product_movie_basic_info 字段英文名字段类型描述是否是分区列 movie_ name STRING 电影名称 dirctor STRING 导演 scriptwriter STRING 编剧 area STRING 制片地区/国家 actors STRING 主演 type STRING 类型 movie_length STRING 电影长度 movie_date STRING 上映日期 movie_language STRING 语言 imdb_ url STRING imdb号 ds STRING 日期分区列数据样例 Select * from public_data.dwd_product_movie_basic_info where ds ='20170119' limit 10; 票房基本信息 ods_product_movie_box 字段英文名字段类型描述是否是分区列 rank STRING 排名 avgprice STRING 平均票价 avppeople STRING 场均人次 boxoffice STRING 单日票房（万） boxoffice_ up STRING 环比变化（%） irank STRING 排名 movieday STRING 上映天数 moviename STRING 影片名 sumboxoffice STRING 累计票房（万） womindex STRING 口碑指数 ds STRING 日期分区列数据样例 Select * from public_data.ods_product_movie_box where ds ='20170119' limit 10; 目标分析需要统计出每个国家/地区累计票房（万）最高的10个电影并且展现这些电影的导演和主演包括的字段是 [ 制片地区/国家、电影名称、排名、累计票房（万）、导演、主演] 现状分析电影基本信息表中包含了 [ 制片地区/国家、电影名称、导演、主演 ] -票房基本信息表中包含了 [ 影片名、累计票房（万）], 其中的 [ 排名 ] 字段不是票房排名 -两个表中没有类似于 [ 影片ID ] 的字段数据源中的数据每天都进行全量更新实现路径 1 创建目标表 2 创建任务，通过 [ 电影名称 ] 和 [ 影片名 ] 建立两张表的连接，并按 [ 制片地区/国家 ] 进行分组，对 [ 累计票房（万）] 进行排名 3 配置调度，该任务每天调度一次实现步骤 1 创建目标表在大数据开发套件的数据开发标签的工具栏中，点击新建，在下拉菜单中选择新建表输入建表语句 CREATE TABLE IF NOT EXISTS movie_sumboxoffice_top10 ( area STRING, movie_name STRING, movie_rank STRING, sumboxoffice STRING, dirctor STRING, actors STRING ) PARTITIONED BY ( ds STRING ); 点击确定提交 2 创建查询任务进行分析这次在工具栏中选择新建任务，选择节点任务，ODPS_SQL类型和周期调度编辑SQL代码，先测试一下。 --连接两张原表 --每天任务运行时都是取当天数据源的最新分区 DROP TABLE IF EXISTS t_movie_sumboxoffice_info; CREATE TABLE IF NOT EXISTS t_movie_sumboxoffice_info AS SELECT a.area , a.movie_name , b.sumboxoffice , a.dirctor , a.actors FROM public_data.dwd_product_movie_basic_info a JOIN (SELECT * FROM public_data.ods_product_movie_box WHERE ds = '${bdp.system.bizdate}')b ON a.movie_name = b.moviename WHERE a.ds = '${bdp.system.bizdate}'; 发现产生的表 t_movie_sumboxoffice_info是空的，原因是数据源的两张表里面并不是每一天都有数据的，做一下数据探查。 SELECT DISTINCT a.ds AS dt FROM public_data.dwd_product_movie_basic_info a JOIN public_data.ods_product_movie_box b ON a.ds = b.ds AND b.ds IS NOT NULL WHERE a.ds IS NOT NULL ORDER BY dt DESC LIMIT 365; 注意order by必须与limit 联用，且order bykey必须是select语句的输出列，即列的别dt。当两个表进行join的时候，主表的Where限制可以写在最后，但从表分区限制条件要写在ON条件或者子查询，否则会先Join后进行分区裁切，造成不必要的性能下降。得到的结果如下（截图中只显示了一部分）最大日期是2017年6月30日，最小日期是2017年1月13日，其中还有不连续的现象。所以现在看来每天有新的分析报告是不行了，那就对历史数据进行一下分析。需要调整一下系统参数，设置为其中较早的一天，比如2017年1月19日。再次运行，得到的结果如下看来是数据没有处理干净，存在属于同一部电影但主演名字不一样的情况。虽然这属于数据清洗的范畴，但是可以简单进行一下去重处理。 --用distinct和wm_concat聚合函数进行简单去重 DROP TABLE IF EXISTS tmp_movie_sumboxoffice_info; CREATE TABLE IF NOT EXISTS tmp_movie_sumboxoffice_info AS SELECT a.area , a.movie_name , a.sumboxoffice , a.dirctor , wm_concat(',', a.actors) as actors FROM ( SELECT distinct area , movie_name , sumboxoffice , dirctor , actors FROM t_movie_sumboxoffice_info ) a group by a.area , a.movie_name , a.sumboxoffice , a.dirctor; 去重后结果如下现在可以继续编辑第三段SQL代码 --产出最终目标表：每个国家/地区总票房最高的10个电影，并展示导演和主演 --每天任务运行产出的日期分区值与源表数据日期一致 INSERT OVERWRITE TABLE movie_sumboxoffice_top10 PARTITION (ds = '${bdp.system.bizdate}') SELECT a.area , a.movie_name , a.movie_rank , a.sumboxoffice , a.dirctor , a.actors FROM ( SELECT area , movie_name , ROW_NUMBER() OVER (PARTITION BY area ORDER BY sumboxoffice DESC) AS movie_rank , sumboxoffice , dirctor , actors FROM tmp_movie_sumboxoffice_info ) a WHERE a.movie_rank < 11; 这里用一个窗口函数Row_number 对 area 开窗, 按照 sumboxoffice 从大到小进行排名最终得到结果如下只有这么可怜的几条数据，选的日子不够好。而且为什么排名第二的票房是6291？感觉似乎少了1个0，感兴趣的同学可以研究一下为什么，或者留言指出我的错误。3 配置调度虽然数据不完整，但是还是可以配置一下调度由于本例中没有依赖其他任务，所以只需要配置调度周期，而且由于是每天调度一次，直接保留默认配置即可。总结整个任务基本上完成，虽然数据不太完整。从去年接触MaxCompute到现在，虽然使用的频率还不够多，但是能感受到功能越来越完善。比如费用预估就是一个非常赞的功能，用户体验大大上升，跑代码的时候心理压力小了很多。公司的OLTP目前是建立在SQL Server 上, 未来可能有迁移数据仓库到MaxCompute的打算。如果有机会的话，到时候再来跟大家分享，包括一些MaxCompute更高级的功能。

2017-07-29

hbase优化实践

hbase优化一：垃圾回收优化： region服务器处理过大的负载，内存分配策略无法安全地只依赖JRE对程序的行为的各种假设，需要使用JRE提供的选项调整垃圾回收策略应对。写入磁盘的数据客户端不连续，导致Java虚拟机堆内存出现空洞。年轻代空间：128~512M之间老生代：好几G。配置文件添加： hbase-env.sh： HBASEOPTS或者HBASEREGIONSERVER_OPT(推荐) 推荐配置： exportHBASE_REGIONOBSERVER_OPTS=" -Xmx8g \ -Xms8g \ -Xmn128m\ -XX:+UseParNewGC\ -XX:+UseConcMarkSweepGC \ -XX:CMSInitiatingOccupancyFraction=70 \ -verbose:gc \ -XX:+PrintGCDetails\ -XX:+PrintGCTimeStamps \ -Xloggc:$HBASE_HOME/logs/gc-${HOSTNAME}-hbase.log" 参照： http://blog.csdn.net/kthq/article/details/8618052 http://swcdxd.iteye.com/blog/1859858 二：hbase压缩可用编码器：GZIP/LZO/Snappy Snappy性能稍好，多使用Snappy hbase启动检查压缩： hbase.regionserver.codecs snappy,lzo 启用压缩： hbase> create 'test2', { NAME => 'cf2', COMPRESSION => 'SNAPPY' } hbase> describe 'test' DESCRIPTION ENABLED 'test', {NAME => 'cf', DATA_BLOCK_ENCODING => 'NONE false ', BLOOMFILTER => 'ROW', REPLICATION_SCOPE => '0', VERSIONS => '1', COMPRESSION => 'GZ', MIN_VERSIONS => '0', TTL => 'FOREVER', KEEP_DELETED_CELLS => 'fa lse', BLOCKSIZE => '65536', IN_MEMORY => 'false', BLOCKCACHE => 'true'} 1 row(s) in 0.1070 seconds 或者：hbase> disable 'test' hbase> alter 'test', {NAME => 'cf', COMPRESSION => 'GZ'} hbase> enable 'test' 三：优化拆分与合并 3.1管理拆分 hbase可能出现‘拆分/合并风暴’ 关闭自动管理拆分，启用手动 To disable automatic splitting, set hbase.hregion.max.filesize to a very large value, such as 100 GB It is not recommended to set it to its absolute maximum value of Long.MAX_VALUE. 3.2 region热点问题 /rowkey的设计一：salting前缀设计/ byte prefix = (byte) (Long.hashCode(System.currentTimeMillis()) % 8); byte[] rowkey1 = Bytes.add(Bytes.toBytes(prefix), Bytes.toBytes(System.currentTimeMillis())); /rowkey的设计二：字段交换，提升权重/ value + System.currentTimeMillis(); /rowkey的设计三：随机化/ MessageDigest md = MessageDigest.getInstance("MD5"); byte[] rowkey3 = md.digest(Bytes.toBytes(System.currentTimeMillis())); /rowkey的设计四：时间顺序/ long rowkey4 = Long.MAX_VALUE - System.currentTimeMillis(); 还可以使用API中move（）region移动到另一个regionserver；或者UNassign移除受影响的表的region 3.3预拆分region 创建表指定需要的region数目 hbase>create 't1','f',SPLITS => ['10','20',30'] hbase>create 't14','f',SPLITS_FILE=>'splits.txt' # create table with four regions based on random bytes keys hbase>create 't2','f1', { NUMREGIONS => 4 , SPLITALGO => 'UniformSplit' } # create table with five regions based on hex keys hbase>create 't3','f1', { NUMREGIONS => 5, SPLITALGO => 'HexStringSplit' } 参考：http://hbase.apache.org/book.html#compression 四：负载均衡： Use the shell to disable the balancer: hbase(main):001:0> balance_switch false true 0 row(s) in 0.3590 seconds This turns the balancer OFF. To reenable, do: hbase(main):001:0> balance_switch true false 0 row(s) in 0.3590 seconds 五：合并region：某些特出情况下，用户需要合并region（删除了大量数据） $ bin/hbase org.apache.hadoop.hbase.util.Merge （If you feel you have too many regions and want to consolidate them, Merge is the utility you need. Merge must run be done when the cluster is down）六：客户端api优化： 6.1禁止自动刷写有大量的写入操作 When performing a lot of Puts, make sure that setAutoFlush is set to false on your Table instance. Otherwise, the Puts will be sent one at a time to the RegionServer. Puts added via table.add(Put) and table.add( Put) wind up in the same write buffer. If autoFlush = false, these messages are not sent until the write-buffer is filled. To explicitly flush the messages, call flushCommits. Calling close on the Table instance will invoke flushCommits. 6.2使用扫描缓存比如：hbase作为mapreduce输入源。设置setCaching比默认值大多的值。 If HBase is used as an input source for a MapReduce job, for example, make sure that the input Scan instance to the MapReduce job has setCaching set to something greater than the default (which is 1). Using the default value means that the map-task will make call back to the region-server for every record processed. Setting this value to 500, for example, will transfer 500 rows at a time to the client to be processed 6.3限定扫描范围 6.4关闭resultScanner 七：配置优化； 7.1减少zookeeper超时 zookeeper.session.timeout 默认三分钟 7.2增加regionserver处理线程 hbase.regionserver.handler.count 默认10 7.3增加region大小管理较少的region可以集群运行更平稳默认256M 7.4减少最大日志文件数目对于写压力比较大的应用，降低值强迫服务器频繁将数据写到磁盘，刷写到磁盘的数据的日志就可以丢弃了。 7.5启用数据压缩

2017-05-08

人脸识别实践

前言很多人都认为人脸识别是一项非常难以实现的工作，看到名字就害怕，然后心怀忐忑到网上一搜，看到网上N页的教程立马就放弃了。这些人里包括曾经的我自己。其实如果如果你不是非要深究其中的原理，只是要实现这一工作的话，人脸识别也没那么难。今天我们就来看看如何在40行代码以内简单地实现人脸识别。一点区分对于大部分人来说，区分人脸检测和人脸识别完全不是问题。但是网上有很多教程有意无意地把人脸检测说成是人脸识别，误导群众，造成一些人认为二者是相同的。其实，人脸检测解决的问题是确定一张图上有木有人脸，而人脸识别解决的问题是这个脸是谁的。可以说人脸检测是是人识别的前期工作。今天我们要做的是人脸识别。所用工具 Anaconda 2 —— Python 2 Dlib scikit-image Dlib 对于今天要用到的主要工具，还是有必要多说几句的。Dlib是基于现代C++的一个跨平台通用的框架，作者非常勤奋，一直在保持更新。Dlib内容涵盖机器学习、图像处理、数值算法、数据压缩等等，涉猎甚广。更重要的是，Dlib的文档非常完善，例子非常丰富。就像很多库一样，Dlib也提供了Python的接口，安装非常简单，用pip只需要一句即可： pip install dlib 上面需要用到的scikit-image同样只是需要这么一句： pip install scikit-image 注：如果用pip install dlib安装失败的话，那安装起来就比较麻烦了。错误提示很详细，按照错误提示一步步走就行了。人脸识别之所以用 Dlib 来实现人脸识别，是因为它已经替我们做好了绝大部分的工作，我们只需要去调用就行了。Dlib里面有人脸检测器，有训练好的人脸关键点检测器，也有训练好的人脸识别模型。今天我们主要目的是实现，而不是深究原理。感兴趣的同学可以到官网查看源码以及实现的参考文献。今天的例子既然代码不超过40行，其实是没啥难度的。有难度的东西都在源码和论文里。首先先通过文件树看一下今天需要用到的东西：准备了六个候选人的图片放在candidate-faces文件夹中，然后需要识别的人脸图片test.jpg。我们的工作就是要检测到test.jpg中的人脸，然后判断她到底是候选人中的谁。另外的girl-face-rec.py是我们的python脚本。shape_predictor_68_face_landmarks.dat是已经训练好的人脸关键点检测器。dlib_face_recognition_resnet_model_v1.dat是训练好的ResNet人脸识别模型。ResNet是何凯明在微软的时候提出的深度残差网络，获得了 ImageNet 2015 冠军，通过让网络对残差进行学习，在深度和精度上做到了比 CNN 更加强大。 1. 前期准备 shape_predictor_68_face_landmarks.dat和dlib_face_recognition_resnet_model_v1.dat都可以在这里找到。不能点击超链接的可以直接输入以下网址：http://dlib.net/files/。然后准备几个人的人脸图片作为候选人脸，最好是正脸。放到candidate-faces文件夹中。本文这里准备的是六张图片，如下：她们分别是然后准备四张需要识别的人脸图像，其实一张就够了，这里只是要看看不同的情况：可以看到前两张和候选文件中的本人看起来还是差别不小的，第三张是候选人中的原图，第四张图片微微侧脸，而且右侧有阴影。 2.识别流程数据准备完毕，接下来就是代码了。识别的大致流程是这样的：先对候选人进行人脸检测、关键点提取、描述子生成后，把候选人描述子保存起来。然后对测试人脸进行人脸检测、关键点提取、描述子生成。最后求测试图像人脸描述子和候选人脸描述子之间的欧氏距离，距离最小者判定为同一个人。 3.代码代码不做过多解释，因为已经注释的非常完善了。以下是girl-face-rec.py # -*- coding: UTF-8 -*- import sys,os,dlib,glob,numpy from skimage import io if len(sys.argv) != 5: print "请检查参数是否正确" exit() # 1.人脸关键点检测器 predictor_path = sys.argv[1] # 2.人脸识别模型 face_rec_model_path = sys.argv[2] # 3.候选人脸文件夹 faces_folder_path = sys.argv[3] # 4.需识别的人脸 img_path = sys.argv[4] # 1.加载正脸检测器 detector = dlib.get_frontal_face_detector() # 2.加载人脸关键点检测器 sp = dlib.shape_predictor(predictor_path) # 3. 加载人脸识别模型 facerec = dlib.face_recognition_model_v1(face_rec_model_path) # win = dlib.image_window() # 候选人脸描述子list descriptors = [] # 对文件夹下的每一个人脸进行: # 1.人脸检测 # 2.关键点检测 # 3.描述子提取 for f in glob.glob(os.path.join(faces_folder_path, "*.jpg")): print("Processing file: {}".format(f)) img = io.imread(f) #win.clear_overlay() #win.set_image(img) # 1.人脸检测 dets = detector(img, 1) print("Number of faces detected: {}".format(len(dets))) for k, d in enumerate(dets): # 2.关键点检测 shape = sp(img, d) # 画出人脸区域和和关键点 # win.clear_overlay() # win.add_overlay(d) # win.add_overlay(shape) # 3.描述子提取，128D向量 face_descriptor = facerec.compute_face_descriptor(img, shape) # 转换为numpy array v = numpy.array(face_descriptor) descriptors.append(v) # 对需识别人脸进行同样处理 # 提取描述子，不再注释 img = io.imread(img_path) dets = detector(img, 1) dist = [] for k, d in enumerate(dets): shape = sp(img, d) face_descriptor = facerec.compute_face_descriptor(img, shape) d_test = numpy.array(face_descriptor) # 计算欧式距离 for i in descriptors: dist_ = numpy.linalg.norm(i-d_test) dist.append(dist_) # 候选人名单 candidate = ['Unknown1','Unknown2','Shishi','Unknown4','Bingbing','Feifei'] # 候选人和距离组成一个dict c_d = dict(zip(candidate,dist)) cd_sorted = sorted(c_d.iteritems(), key=lambda d:d[1]) print "\n The person is: ",cd_sorted[0][0] dlib.hit_enter_to_continue() 4.运行结果我们在.py所在的文件夹下打开命令行，运行如下命令 python girl-face-rec.py 1.dat 2.dat ./candidate-faecs test1.jpg 由于shape_predictor_68_face_landmarks.dat和dlib_face_recognition_resnet_model_v1.dat名字实在太长，所以我把它们重命名为1.dat和2.dat。运行结果如下： The person is Bingbing。记忆力不好的同学可以翻上去看看test1.jpg是谁的图片。有兴趣的话可以把四张测试图片都运行下试试。这里需要说明的是，前三张图输出结果都是非常理想的。但是第四张测试图片的输出结果是候选人4。对比一下两张图片可以很容易发现混淆的原因。机器毕竟不是人，机器的智能还需要人来提升。有兴趣的同学可以继续深入研究如何提升识别的准确率。比如每个人的候选图片用多张，然后对比和每个人距离的平均值之类的。全凭自己了。

2017-04-16

优酷鸿蒙开发实践｜多屏互动开发实践

鸿蒙系统初实践普通流转 “鸿蒙流转” 指利用了鸿蒙操作系统特性的多屏互动方式，与传统投屏协议最大的区别在于对鸿蒙操作系统的依赖。

2021-10-27

精选列表