TraceApiAnalyzeMinutesSink.java 3.63 KB
package com.yoho.trace.online.sink;

import com.alibaba.fastjson.JSONObject;
import com.yoho.trace.anaylzer.model.ApiTraceResult;
import com.yoho.trace.store.HBasePool;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;

/**
 * Created by mingdan.ge on 2019/11/11.
 */
public class TraceApiAnalyzeMinutesSink extends RichSinkFunction<Tuple2<String, ApiTraceResult>> {
    private Connection conn;
    private HTable table;

    @Override
    public void open(Configuration parameters) throws Exception {
        conn = HBasePool.getConnection();
        table = (HTable) conn.getTable(TableName.valueOf("trace_api_analyze_minutes"));
        table.setWriteBufferSize(1024 * 1024 * 20);
        table.setAutoFlush(false, true);//不单个提交
    }

    @Override
    public void invoke(Tuple2<String, ApiTraceResult> value, SinkFunction.Context context) throws Exception {
        long now = System.currentTimeMillis()/1000;
        ApiTraceResult apiTraceResult = value.f1;
        String[] md5Tags = StringUtils.split(apiTraceResult.getTraceMd5(), '.');
        String md5 = "";

        if (md5Tags!= null && md5Tags.length==2) {
            md5 = md5Tags[1];
        }

        String rowkey = apiTraceResult.getApiName() + ":" + now + ":" + md5;
        Put put = new Put(Bytes.toBytes(rowkey));
        put.addColumn(Bytes.toBytes("trace"), Bytes.toBytes("api"), Bytes.toBytes(apiTraceResult.getApiName()));
        put.addColumn(Bytes.toBytes("trace"), Bytes.toBytes("traceMd5"), Bytes.toBytes(md5));
        put.addColumn(Bytes.toBytes("trace"), Bytes.toBytes("duration"), Bytes.toBytes(apiTraceResult.getDuration()));
        put.addColumn(Bytes.toBytes("trace"), Bytes.toBytes("times"), Bytes.toBytes(apiTraceResult.getCallTimes()));
        put.addColumn(Bytes.toBytes("trace"), Bytes.toBytes("maxLatency"), Bytes.toBytes(apiTraceResult.getMaxLatency()));
        put.addColumn(Bytes.toBytes("trace"), Bytes.toBytes("maxLatencyTrace"), Bytes.toBytes(apiTraceResult.getMaxLatencyTrace()));
        put.addColumn(Bytes.toBytes("trace"), Bytes.toBytes("minLatency"), Bytes.toBytes(apiTraceResult.getMinLatency()));
        put.addColumn(Bytes.toBytes("trace"), Bytes.toBytes("minLatencyTrace"), Bytes.toBytes(apiTraceResult.getMinLatencyTrace()));
        put.addColumn(Bytes.toBytes("trace"), Bytes.toBytes("spans"), Bytes.toBytes(JSONObject.toJSONString(apiTraceResult.getSpans())));
        put.addColumn(Bytes.toBytes("trace"), Bytes.toBytes("restTemplateTimes"), Bytes.toBytes(apiTraceResult.getRestTemplateTimes()));
        put.addColumn(Bytes.toBytes("trace"), Bytes.toBytes("redisTimes"), Bytes.toBytes(apiTraceResult.getRedisTimes()));
        put.addColumn(Bytes.toBytes("trace"), Bytes.toBytes("mysqlTimes"), Bytes.toBytes(apiTraceResult.getMysqlTimes()));
        if(StringUtils.isNotEmpty(apiTraceResult.getRegion())){
            put.addColumn(Bytes.toBytes("trace"), Bytes.toBytes("region"), Bytes.toBytes(apiTraceResult.getRegion()));
        }
        table.put(put);
    }

    @Override
    public void close() throws Exception {
        try {
            table.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        super.close();
        conn.close();
    }
}