mycat系列-Mycat的join

分片join

join概述

Join绝对是关系型数据库中最常用一个特性，然而在分布式环境中,跨分片的join确是最复杂的，最难解决一个问题。

下面我们简单介绍下各种Join操作。

INNER JOIN

内连接，也叫等值连接，inner join产生同时符合A表和B表的一组数据。

如图：

LEFT JOIN

左连接从A表(左)产生一套完整的记录,与匹配的B表记录(右表) .如果没有匹配,右侧将包含null,在Mysql中等同于left outer join。

如图：

RIGHT JOIN

同Left join,AB表互换即可。

Cross join

交叉连接，得到的结果是两个表的乘积，即笛卡尔积。笛卡尔（Descartes）乘积又叫直积。假设集合A={a,b}，集合B={0,1,2}，则两个集合的笛卡尔积为{(a,0),(a,1),(a,2),(b,0),(b,1), (b,2)}。可以扩展到多个集合的情况。类似的例子有，如果A表示某学校学生的集合，B表示该学校所有课程的集合，则A与B的笛卡尔积表示所有可能的选课情况。

Full join

全连接产生的所有记录（双方匹配记录）在表A和表B。如果没有匹配,则对面将包含null。

性能建议

尽量避免使用Left join或Right join,而用Inner join

在使用Left join或Right join时，ON会优先执行，where条件在最后执行，所以在使用过程中，条件尽可能的在ON语句中判断，减少where的执行

少用子查询，而用join。

Mycat目前版本支持跨分片的join,主要实现的方式有四种。

全局表，ER分片，catletT(人工智能)和ShareJoin，

ShareJoin在开发版中支持，前面三种方式1.3.0.1支持。

全局表

一个真实的业务系统中，往往存在大量的类似字典表的表格，它们与业务表之间可能有关系，这种关系，可以理解为“标签”，而不应理解为通常的“主从关系”，这些表基本上很少变动，可以根据主键ID进行缓存，下面这张图说明了一个典型的“标签关系”图：

在分片的情况下，当业务表因为规模而进行分片以后，业务表与这些附属的字典表之间的关联，就成了比较棘手的问题，考虑到字典表具有以下几个特性：

•  变动不频繁

•  数据量总体变化不大

•  数据规模不大，很少有超过数十万条记录。

鉴于此，MyCAT定义了一种特殊的表，称之为“全局表”，全局表具有以下特性：

•  全局表的插入、更新操作会实时在所有节点上执行，保持各个分片的数据一致性

•  全局表的查询操作，只从一个节点获取

•  全局表可以跟任何一个表进行JOIN操作

将字典表或者符合字典表特性的一些表定义为全局表，则从另外一个方面，很好的解决了数据JOIN的难题。通过全局表+基于E-R关系的分片策略，MyCAT可以满足80%以上的企业应用开发。

配置

全局表配置比较简单，不用写Rule规则，如下配置即可：

<table name="company" primaryKey="ID" type="global" dataNode="dn1,dn2,dn3" />

需要注意的是，全局表每个分片节点上都要有运行创建表的DDL语句。 

ER Join

MyCAT借鉴了NewSQL领域的新秀Foundation DB的设计思路，Foundation DB创新性的提出了Table Group的概念，其将子表的存储位置依赖于主表，并且物理上紧邻存放，因此彻底解决了JION的效率和性能问题，根据这一思路，提出了基于E-R关系的数据分片策略，子表的记录与所关联的父表记录存放在同一个数据分片上。

customer采用sharding-by-intfile这个分片策略，分片在dn1,dn2上，orders依赖父表进行分片，两个表的关联关系为orders.customer_id=customer.id。于是数据分片和存储的示意图如下：

这样一来，分片Dn1上的的customer与Dn1上的orders就可以进行局部的JOIN联合，Dn2上也如此，再合并两个节点的数据即可完成整体的JOIN，试想一下，每个分片上orders表有100万条，则10个分片就有1个亿，基于E-R映射的数据分片模式，基本上解决了80%以上的企业应用所面临的问题。

配置

以上述例子为例，schema.xml中定义如下的分片配置：

<table name="customer" dataNode="dn1,dn2" rule="sharding-by-intfile"> <childTable name="orders"  joinKey="customer_id" parentKey="id"/></table>

Share join

ShareJoin是一个简单的跨分片Join,基于HBT的方式实现。

目前支持2个表的join,原理就是解析SQL语句，拆分成单表的SQL语句执行，然后把各个节点的数据汇集。

配置

支持任意配置的A,B表如：

A,B的dataNode相同

<table name="A" dataNode="dn1,dn2,dn3" rule="auto-sharding-long" /><table name="B" dataNode="dn1,dn2,dn3" rule="auto-sharding-long" />

A,B的dataNode不同 

<table name="A" dataNode="dn1,dn2 " rule="auto-sharding-long" /><table name="B" dataNode="dn1,dn2,dn3" rule="auto-sharding-long" />

或 

<table name="A" dataNode="dn1 " rule="auto-sharding-long" /><table name="B" dataNode=" dn2,dn3" rule="auto-sharding-long" />

代码测试

先把表company从全局表修改下配置

<table name="company" primaryKey="ID"  dataNode="dn1,dn2,dn3"  rule="mod-long" />

重新插入数据 

mysql> delete from company;Query OK, 9 rows affected (0.19 sec)mysql> insert company (id,name) values(1,'mycat');Query OK, 1 row affected (0.08 sec)mysql> insert company (id,name) values(2,'ibm');Query OK, 1 row affected (0.03 sec)mysql> insert company (id,name) values(3,'hp');Query OK, 1 row affected (0.03 sec)

下面可以看下普通的join和sharejoin的区别 

mysql> select a.*,b.id, b.name as tit from customer a,company b where a.company_id=b.id;+----+------+------------+-------------+----+------+| id | name | company_id | sharding_id | id | tit  |+----+------+------------+-------------+----+------+|  3 | feng |          3 |       10000 |  3 | hp   |+----+------+------------+-------------+----+------+1 row in set (0.03 sec)mysql> /*!mycat:catlet=demo.catlets.ShareJoin */ select a.*,b.id, b.name as tit from customer a,company b on a.company_id=b.id;+----+------+------------+-------------+----+-------+| id | name | company_id | sharding_id | id | tit   |+----+------+------------+-------------+----+-------+|  3 | feng |          3 |       10000 |  3 | hp    ||  1 | wang |          1 |       10000 |  1 | mycat ||  2 | xue  |          2 |       10010 |  2 | ibm   |+----+------+------------+-------------+----+-------+3 rows in set (0.05 sec)

其他两种写法 

/*!mycat:catlet=demo.catlets.ShareJoin */ select a.*,b.id, b.name as tit from customer a join company b on a.company_id=b.id;+----+------+------------+-------------+----+-------+| id | name | company_id | sharding_id | id | tit   |+----+------+------------+-------------+----+-------+|  3 | feng |          3 |       10000 |  3 | hp    ||  1 | wang |          1 |       10000 |  1 | mycat ||  2 | xue  |          2 |       10010 |  2 | ibm   |+----+------+------------+-------------+----+-------+3 rows in set (0.01 sec)/*!mycat:catlet=demo.catlets.ShareJoin */ select a.*,b.id, b.name as tit from customer a join company b where a.company_id=b.id;+----+------+------------+-------------+----+-------+| id | name | company_id | sharding_id | id | tit   |+----+------+------------+-------------+----+-------+|  3 | feng |          3 |       10000 |  3 | hp    ||  1 | wang |          1 |       10000 |  1 | mycat ||  2 | xue  |          2 |       10010 |  2 | ibm   |+----+------+------------+-------------+----+-------+3 rows in set (0.01 sec)

对*的支持，还可以这样写SQL 

mysql> /*!mycat:catlet=demo.catlets.ShareJoin */ select a.*,b.* from customer a join company b on a.company_id=b.id;+----+------+------------+-------------+-------+| id | name | company_id | sharding_id | name  |+----+------+------------+-------------+-------+|  1 | wang |          1 |       10000 | mycat ||  2 | xue  |          2 |       10010 | ibm   ||  3 | feng |          3 |       10000 | hp    |+----+------+------------+-------------+-------+3 rows in set (0.02 sec)mysql> /*!mycat:catlet=demo.catlets.ShareJoin */ select * from customer a join company b on a.company_id=b.id;+----+------+------------+-------------+-------+| id | name | company_id | sharding_id | name  |+----+------+------------+-------------+-------+|  1 | wang |          1 |       10000 | mycat ||  2 | xue  |          2 |       10010 | ibm   ||  3 | feng |          3 |       10000 | hp    |+----+------+------------+-------------+-------+3 rows in set (0.02 sec)

/*!mycat:catlet=demo.catlets.ShareJoin */ select a.id,a.user_id,a.traveldate,a.fee,a.days,b.id as nnid, b.title as tit from travelrecord  a  join  hotnews b on b.id=a.days order by a.id ;

catlet（人工智能）

解决跨分片的SQL JOIN的问题，远比想象的复杂，而且往往无法实现高效的处理，既然如此，就依靠人工的智力，去编程解决业务系统中特定几个必须跨分片的SQL的JOIN逻辑，MyCAT提供特定的API供程序员调用，这就是MyCAT创新性的思路——人工智能。

以一个跨节点的SQL为例，

Select a.id,a.name,b.title from a,b where a.id=b.id

其中a在分片1，2，3上，b在4，5，6上，需要把数据全部拉到本地（MyCAT服务器）,执行JOIN逻辑，具体过程如下（只是一种可能的执行逻辑）： 

EngineCtx ctx=new EngineCtx();//包含MyCat.SQLEngineString sql=,“select a.id ,a.name from a ”;

//在a表所在的所有分片上顺序执行下面的本地SQL

ctx.executeNativeSQLSequnceJob(allAnodes,new DirectDBJoinHandler());

DirectDBJoinHandler类是一个回调类，负责处理SQL执行过程中返回的数据包，这里的这个类，主要目的是用a表返回的ID信息，去b表上查询对于的记录，做实时的关联：

DirectDBJoinHandler{  Private HashMap<byte[],byte[]> rows;//Key为id,value为一行记录的Column原始Byte数组，这里是a.id,a.name,b.title这三个要输出的字段   Public Boolean onHeader(byte[] header){  //保存Header信息，用于从Row中获取Field字段值}   Public Boolean onRowData(byte[] rowData){    String id=getColumnAsString(“id”);//放入结果集,b.title字段未知，所以先空着rows.put(getColumnRawBytes(“id”),rowData);  //满1000条，发送一个查询请求String sql=”select b.id, b.name  from b where id in (………….)”;//此SQL在B的所有节点上并发执行，返回的结果直接输出到客户端 ctx.executeNativeSQLParallJob(allBNodes,sql ,new MyRowOutPutDataHandler(rows));}   Public Boolean onRowFinished()  { }Public void onJobFinished() {If(ctx.allJobFinished())     {///used total time ….     }}}

最后，增加一个Job事件监听器，这里是所有Job完成后，往客户端发送RowEnd包，结束整个流程。

ctx.setJobEventListener(new JobEventHandler(){public void onJobFinished(){ client.writeRowEndPackage()}});

以上提供一个SQL执行框架，完全是异步的模式执行，并且以后会提供更多高质量的API，简化分布式数据处理，比如内存结合文件的数据JOIN算法，分组算法，排序算法等等，

期待更多的牛人一起来完善。

Spark/Storm 对join扩展

看到这个标题，可能会感到很奇怪，Spark和Storm 和Join有关系吗? 有必要用Spark，storm吗？

mycat后续的功能会引入spark和storm来做跨分片的join,大致流程是这样的在mycat调用spark,storm的api,把数据传送到spark,storm，

在spark,storm进行join,在把数据传回mycat,mycat在返回给客户端。