1.抽样检测分组字段，看看是否有倾斜
2.如果没有倾斜，就正常增加reduce数量，设置中间ORC+SNAPPY压缩
3.如果有倾斜，把倾斜的部分过滤出来加前缀打散处理，不倾斜的部分正常处理。
4.如果是大表join小表，大表倾斜，可以使用map端join方法，小表倾斜直接无视。
5.如果是大表join大表，某表倾斜，可以使用膨胀法处理。

1 将值域划分区间建有序链表，原本海量的遍历会被压缩N倍。每个值域对应一个有序链表，因此是双重链表。
2 前提：除了跳表外还有一张作品表，以作品id为key，value含分数
3 值变更时，先查询作品表更新原值，通过原值到对应区间的链表中删除该值，新值插入对应区间的链表中。
4 求排名：1.每个区间链表除了保存作品和分数外，还额外维护一个区间总长
5 2.每次新值插入时，都将原区间总长-1，新区间总长+1
6 3.求排名时遍历第一层链表将区间总长累加，然后遍历该值所在的区间求出区间内排名，两者相加。
7        
8 求TopN:第二层链表一路往下走N个。
9 缺点：事实上所有针对数据直接排序的方式都存在锁问题，每次更新只能有一个线程做，否则必然会乱。

1 假设总分为5                 [0,5]
2 1.将总分一分为二            [0,2],[3,5]
3 2.将划分的部分继续二分      [0,1],[1,2],[3,4],[4,5]
4 3.重复2直到划分到底         [0,0],[1,1],[2,2],[3,3]...
5
6 采用链表的方式从顶点往下一路关联起来，每个区间key对应一个人数value。
7 假设有某个值从2变成3。
8 先从顶点开始往下遍历查找[2,2]，遍历过程中经过的节点人数全部-1，终点[2,2]的人数也-1;
9 然后从顶点开始往下遍历查找[3,3]，经过的节点人数全部+1。终点也+1。
10  求排名：假设要求分数为3的排名，从起点开始往下查找[3,3]，如果是往左走，就把右值累加上，如果往右走则不加。
11  优点：毫秒级响应排名，更新不需要锁表，因为是单纯的加减日志。
12  缺点：消耗大量内存存储，每次更新都要改变N个值，每次更新值时还需要把排名添加到TopN表里。

理论：
如果是k-v形式的数据，或者是可以提取key的数据，可以使用分组法比较。
假设我有x G内存，要求A,B的交集。
先把A,B分别分组拆分成多个小文件，每个小文件的大小为(x-300)/2，超过的部分再水平拆分成多个 文件。
然后按分组读入A,B的数据，在内存中比较，求出交集部分。
如果分组太细，就按分组key的哈希进行分区，一次比较多个分组。
为了方便小文件的合并，可以在输出时将数据排序输出，创建索引文件，后续可以采用归并排序的方式 快速合并。
如果不能提取key，则按照hash分组。
实战：假设已经把两张表要求交集的部分字段拼接成"key"
select * from A join B on A.key = B.key

 select * from A full outer join B on A.key = B.key where A.key is null or B.key  null;

/**
* 一个完整的快速排序的方法
* 可以传入任意类型的buffer对其排序后返回
* 可以传入一个比较器对自定义类型排序 *
* @author 孤星魅影
* @param buffer 任意类型的buffer
* @param isASC  是否正序排序
* @param ev     自定义比较器
* @return 返回一个快速排序后的buffer */
def quickSort[T: ClassTag](buffer: mutable.Buffer[T])(implicit isASC: Boolean = true, ev: T => Comparable[T]): mutable.Buffer[T] = {
//为null、为空、长度为1时都返回buffer
if (buffer == null | | buffer.length <= 1)
return buffer
//1.将数组的第一个数head与后面所有的数进行比较，左边放比head小的数，右边放比head大的数 (从小到大)
//2.递归执行拆分，直到数组只剩下1个
//3.按顺序将左、中、右三者拼接起来，完成排序
val (left, right) = buffer.tail.partition(t => {
if (isASC) t.compareTo(buffer.head) < 0
else t.compareTo(buffer.head) > 0
})
quickSort(left) += buffer.head ++= quickSort(right)
}

scala
object{
lazy val obj = new T()
}
java
class T{
private static T obj = null;
private T(){}
//先执行一次null检验，然后上锁再执行一次null检验，防止同时有两个线程通过了第一次null检 验导致异常。
public T getObj()={
if(obj==null){
sysconized(obj){
if(obj == null) obj = new T();
        }
    }
return obj;
 }
}

a,b,c,d = a*256*256*256+b*256*256+c*256+123
ip是256进制，所以a需要*3个256才能保证每个bcd不重复，b,c,d同理，使用位计算效率更高：*256 相当于<<8
提供一段scala代码:
def
ip2long(ip:String)=ip.split(".").reverse.zipWithIndex.map(v=>v._1<<(8*v._2)). reduce(_+_)
可以提取转换后的数字的前三位进行分组查询，效率更高：
ip.groupby(_.substring(0,3))

/**
* 求min - max以内的所有质数
* 只需要判断一个数是否能被自身以外的大于1的质数数整除就可以了
* @return */
def countPrime(min:Int,max:Int):Array[Int]={
val arr = mutable.Set[Int](1)
(min to max).foreach(index=> {
if(!arr.exists(prime=>{
//某个数如果能被除自己和1以外的质数整除，说明该数不是质数          if(prime>1 && index != prime && index % prime == 0) true
else false
}))
arr += index
})
arr.toArray.sorted
}

package third;
import static java.lang.Thread.sleep;
public class SellShaoBing {
private int shaoBingAmount=0;
private boolean isStop = false;
private int maxAmount;
public SellShaoBing(int max){
maxAmount = max;
}
private Thread createProductThread(){
return new Thread(()->{
while (!isStop) {
synchronized (this) {
//留个缓冲，做满之前就开始卖
if (shaoBingAmount > maxAmount - 5) notifyAll(); waitMethod(shaoBingAmount == maxAmount);        sleepMethod();
System.out.println("做烧饼");
shaoBingAmount++;
System.out.println("当前烧饼数量"+shaoBingAmount); }
}
});
}
private Thread createConsumerThread(){
return new Thread(()->{
while (!isStop) {
synchronized (this) {
//留个缓冲，烧饼卖完之间就通知生产                       if (shaoBingAmount < maxAmount /4) notifyAll(); waitMethod(shaoBingAmount == 0);               sleepMethod();
System.out.println("卖烧饼");
shaoBingAmount--;
System.out.println("当前烧饼数量"+shaoBingAmount); }
}
});
}
public void actions(){
Thread product =  createProductThread();
Thread consumer =  createConsumerThread();
consumer.start();
product.start();
try {
sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
}
isStop = true;
}
private synchronized void waitMethod(boolean iswait){ if (iswait) {
try {
wait();
} catch (InterruptedException e) { e.printStackTrace();
Thread.currentThread().interrupt(); }
}
}
private void sleepMethod(){
try {
sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt(); }
}
}

package utils
//卖烧饼 (Java生产者消费者的Scala实现)
case class ShaoBing(var
amount:Int=0,maxAmount:Int=30,isContinue:Boolean=true)
object ShaoBing{
/**
* 烧饼线程模板
* @param isNotify 传入一个唤醒所有线程的判断方法
* @param isWait   传入一个使当前线程等待的方法
* @param doMethod 传入一个增减烧饼数量的方法
* @param shaoBing 传入一个烧饼对象 */
def createThread(isNotify:ShaoBing => Boolean , isWait:ShaoBing => Boolean , doMethod:ShaoBing => Unit)(shaoBing:ShaoBing)={
while (shaoBing.isContinue){
//notify和wait都需要在同步代码块中才能使用，java 的 synchronized(obj) {...} = scala 的 obj.synchronized{...}
shaoBing.synchronized{
if (isNotify(shaoBing)) shaoBing.notifyAll()
//传入等待条件，使线程等待
if (isWait(shaoBing)) shaoBing.wait()
//增加或减少烧饼数量时才上锁
doMethod(shaoBing)
}
println("当前烧饼数量:"+shaoBing.amount)
Thread.sleep(1000)
}
}
//做烧饼
val createProducerThread = createThread(
//做满之前就开始卖
isNotify = shaoBing=>shaoBing.amount > shaoBing.maxAmount-5, //做满之后就等
isWait = shaoBing=>shaoBing.amount == shaoBing.maxAmount, _.amount += 1) _
//卖烧饼
val createConsumerThread = createThread(
//卖到剩四分之一时候提前开始生产烧饼
isNotify = shaoBing=>shaoBing.amount < shaoBing.maxAmount/4, //卖光了就停止卖烧饼
isWait = shaoBing=>shaoBing.amount == 0,
_.amount -= 1) _
//启动生产者和消费者
def main(args:Array[String]):Unit={
val shaoBing = ShaoBing()
val producer = new Thread(new Runnable {override def run(): Unit = createProducerThread(shaoBing)})
val consumer = new Thread(new Runnable {override def run(): Unit = createConsumerThread(shaoBing)})
producer.start()
consumer.start()
}
}