flink內部計算指標的95線-99線等的實現
阿新 • • 發佈：2018-12-24
15年在某電商從0設計了一個通用的API監控系統，當時只是計算了成功率+平均耗時，沒有算75,90,95,99,999,9999線，這次單位需要，所以促使我去思考這個問題，問了單位CAT維護人員，大致瞭解了計算方式，跟我在18年7月份在單位內網BBS發表的文章思路是一致的，所以就直接寫了下面的程式碼
PercentageCalculation.java
package com.ymm.computation.udf.define;import org.apache.flink.table.functions.AggregateFunction;import org.slf4j.Logger;import org.slf4j.LoggerFactory;//批量計算95line類似的資料public class PercentageCalculation extends AggregateFunction<Object, PercentageAccumulator> { /** */
private static final long serialVersionUID = 4009559061130131166L; private static final Logger LOG = LoggerFactory
.getLogger(PercentageCalculation.class); //private static BlockingQueue<PercentageAggregatorContainer> GLOBAL_QUEUE = new LinkedBlockingQueue<PercentageAggregatorContainer>();
@Override
public PercentageAccumulator createAccumulator() { return new PercentageAccumulator();
} public void accumulate(PercentageAccumulator accumulator, Object value) {
accumulator.accumulate(value);
} @Override
public Object getValue(PercentageAccumulator accumulator) { return accumulator.getValue();
} public void resetAccumulator(PercentageAccumulator acc) {
acc = null;//help GC
}
}PercentageAccumulator.java
package com.ymm.computation.udf.define;import org.slf4j.Logger;import org.slf4j.LoggerFactory;//!只針對時間來計算95線等,其它引數不要使用本類public class PercentageAccumulator { private static final Logger LOG = LoggerFactory
.getLogger(PercentageAccumulator.class); public final static double PERCENT_50 = 0.5; public final static double PERCENT_75 = 0.25; public final static double PERCENT_90 = 0.1; public final static double PERCENT_95 = 0.05; public final static double PERCENT_99 = 0.01; public final static double PERCENT_999 = 0.001; public final static double PERCENT_9999 = 0.0001; public final static int PERCENT_COUNT = 7; private final static int[] SCALE = { //
1, //
2, //
4, //
8, //
16, //
32, //
64, //
128, //
256, //
512, //
1024, //
2048, //
4096, //
8192, //
16384, //
32768, //
65536 //
}; private int[] countContainer = { //
0, //<=1
0, //<=2
0, //<=4
0, //<=8
0, //<=16
0, //<=32
0, //<=64
0, //<=128
0, //<=256
0, //<=512
0, //<=1024
0, //<=2048
0, //<=4096
0, //<=8192
0, //<=16384
0, //<=32768
0 //<=65536
}; private int positionByTwoDivision(int[] array, int begin, int end, int value) { int mid = (begin + end) >> 1; int midValue = array[mid]; int halfMidValue = midValue >> 1; //判斷是否可以命中mid
if (value > halfMidValue && value <= midValue) { return mid;
} //沒法命中,則根據大小來定
if (value <= halfMidValue) { if (mid - 1 < 0) {//沒路可走的邊界條件
return 0;
} return positionByTwoDivision(array, begin, mid - 1, value);
} else { return positionByTwoDivision(array, mid + 1, end, value);
}
} public int positionInValueArray(int val) { int length = SCALE.length; //如果大於最大值|小於等於最小值
if (val >= SCALE[length - 1]) { return length - 1;
} else if (val <= SCALE[0]) { return 0;
} //採用2分法來計算
return positionByTwoDivision(SCALE, 0, length - 1, val);
} public void accumulate(Object value) { //轉換為long值,int值夠用了
Long longValue = (Long) value; int intValue = longValue.intValue(); //找到下標
int index = positionInValueArray(intValue);
countContainer[index]++;
} //確保在[1,MAX]範圍內,
//自然順序
private int adjust(int input, int max) { if (input <= 1) { return 1;
} else if (input >= max) { return max;
} else { return input;
}
} private static final ThreadLocal<StringBuilder> STR_BUILDER_ThreadLocal = new ThreadLocal<StringBuilder>() { public StringBuilder initialValue() { return new StringBuilder();
}
}; private static final String SEPARATOR = ":"; public String getValue() { //total
int total = 0; int length = countContainer.length; for (int index = 0; index < length; index++) {
total += countContainer[index];
} //如果total為0的異常情況
//注意是自然序---[1,total]
int percent_9999_pos = adjust((int) (total * PERCENT_9999), total); boolean found_9999 = false; int percent_9999_value = Integer.MAX_VALUE; //999
int percent_999_pos = adjust((int) (total * PERCENT_999), total); boolean found_999 = false; int percent_999_value = Integer.MAX_VALUE; //99
int percent_99_pos = adjust((int) (total * PERCENT_99), total); boolean found_99 = false; int percent_99_value = Integer.MAX_VALUE; //95
int percent_95_pos = adjust((int) (total * PERCENT_95), total); boolean found_95 = false; int percent_95_value = Integer.MAX_VALUE; //90
int percent_90_pos = adjust((int) (total * PERCENT_90), total); boolean found_90 = false; int percent_90_value = Integer.MAX_VALUE; //75
int percent_75_pos = adjust((int) (total * PERCENT_75), total); boolean found_75 = false; int percent_75_value = Integer.MAX_VALUE; //50
int percent_50_pos = adjust((int) (total * PERCENT_50), total); boolean found_50 = false; int percent_50_value = Integer.MAX_VALUE; //開始遍歷每一個元素,從後往前算
int scanned = 0; int left = PERCENT_COUNT; for (int index = length - 1; index >= 0; index--) { //當前沒有值,無論如何也不會成為備選
if (0 == countContainer[index]) { continue;
} //當前有值
scanned += countContainer[index]; //逐個判斷
//9999線
if (false == found_9999 && scanned >= percent_9999_pos) {
percent_9999_value = SCALE[index];
found_9999 = true;
left--;
} //999線
if (false == found_999 && scanned >= percent_999_pos) {
percent_999_value = SCALE[index];
found_999 = true;
left--;
} //99線
if (false == found_99 && scanned >= percent_99_pos) {
percent_99_value = SCALE[index];
found_99 = true;
left--;
} //95線
if (false == found_95 && scanned >= percent_95_pos) {
percent_95_value = SCALE[index];
found_95 = true;
left--;
} //90線
if (false == found_90 && scanned >= percent_90_pos) {
percent_90_value = SCALE[index];
found_90 = true;
left--;
} //75線
if (false == found_75 && scanned >= percent_75_pos) {
percent_75_value = SCALE[index];
found_75 = true;
left--;
} //50線
if (false == found_50 && scanned >= percent_50_pos) {
percent_50_value = SCALE[index];
found_50 = true;
left--;
} //全部都找到了就break
if (0 == left) { break;
}
} //所有的值都算好了
//拿出來時先reset一下
StringBuilder stringBuilder = STR_BUILDER_ThreadLocal.get();
stringBuilder.delete(0, stringBuilder.length()); //開始掛各種資料,測試表明每秒幾百萬次執行
stringBuilder.append(percent_50_value);
stringBuilder.append(SEPARATOR);
stringBuilder.append(percent_75_value);
stringBuilder.append(SEPARATOR);
stringBuilder.append(percent_90_value);
stringBuilder.append(SEPARATOR);
stringBuilder.append(percent_95_value);
stringBuilder.append(SEPARATOR);
stringBuilder.append(percent_99_value);
stringBuilder.append(SEPARATOR);
stringBuilder.append(percent_999_value);
stringBuilder.append(SEPARATOR);
stringBuilder.append(percent_9999_value); //return
return stringBuilder.toString();
} public void print() { for (int index = 0; index < this.countContainer.length; index++) {
System.out.println(index + "->" + this.countContainer[index]);
}
}
}歡迎提出 優化建議，比如對GC更友好的優化建議！
這個函式的瓶頸在於stringbuilder.