flink sink 批量下层数据时 ，数据"丢失"？

本文主要是介绍flink sink 批量下层数据时 ，数据"丢失"？，对大家解决编程问题具有一定的参考价值，需要的程序猿们随着小编来一起学习吧！

flink作为目前很火的大数据框架，在实时计算和离线计算场景下有大量应用

图片来源{https://ci.apache.org/projects/flink/flink-docs-release-1.11/learn-flink/}

从图中可以看出，flink在处理数据中有三大块：source operator sink

在工作时负责sink这块的数据下层工作，因为下层源有很多，有redis，hbase这些。在实现hbase相关的sink时，我优先考虑使用批量进行sink下层，这样可以提高下层数据时的效率问题。

最开始的编写hbase相关的sink代码（ 工作相关业务代码已屏蔽）如下：

public class BatchHbaseSink extends RichSinkFunction<JSONObject> {

    private static final Logger logger = LoggerFactory.getLogger(BatchHbaseSink.class);

    private Connection connection;
    private long lastInvokeTime;
    private final List<Put> puts = new ArrayList();
    /**
     * 每次处理最多500条数据
     */
    private final int maxSize = 500;
    /**
     * 最长延迟5秒
     */
    private final long delayTime = 5000L;


    public BatchHbaseSink() {
        System.out.println("new batch =========== " + Thread.currentThread().getName());
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = HBaseUtil.getConnection(
                this.config.getZookeeperQuorum(),
                this.config.getClientPort()
        );
        lastInvokeTime = System.currentTimeMillis();
    }


    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        try {
						.....省略
      
            Put put = new Put(keySuffix + keySuffix).getBytes());
            put.addColumn(family.getBytes(), qualifier.getBytes(), Bytes.toBytes(val.toString()));
            puts.add(put);
            final long currentTime = System.currentTimeMillis();
            final long l = currentTime - lastInvokeTime;
            if (puts.size() >= maxSize || l >= delayTime) {
                Table table = connection.getTable(TableName.valueOf(tableName));
                // 数据提交
                table.put(puts);
                puts.clear();
                lastInvokeTime = currentTime;
                table.close();
            }
        } catch (Exception e) {
            logger.error("error:", e);
        }
    }


    @Override
    public void close() throws IOException {
        if (connection != null) {
            connection.close();
        }
    }
}

我开始的想法很简单，每次等500条数据一次性写入hbase，同时设置延时时间为5秒，防止数据积压

但是当我发送20条需要处理的数据时，发现每次hbase每次写入的数据都不满足20条，运行多次后都有同样的问题，我重新review了代码逻辑，发现逻辑处理方面没有什么问题，我开始把代码改成，来一条数据发送一条数据代码如下

public class BatchHbaseSink extends RichSinkFunction<JSONObject> {
​
    private static final Logger logger = LoggerFactory.getLogger(BatchHbaseSink.class);
​
    private Connection connection;
​
    public BatchHbaseSink() {
        System.out.println("new batch =========== " + Thread.currentThread().getName());
    }
​
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = HBaseUtil.getConnection(
                this.config.getZookeeperQuorum(),
                this.config.getClientPort()
        );
    }
​
​
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        try {
            .....省略
      
            Put put = new Put(keySuffix + keySuffix).getBytes());
            put.addColumn(family.getBytes(), qualifier.getBytes(), Bytes.toBytes(val.toString()));
            Table table = connection.getTable(TableName.valueOf(tableName));
            // 数据提交
            table.put(puts);
            puts.clear();
            lastInvokeTime = currentTime;
            table.close();
         }
        } catch (Exception e) {
            logger.error("error:", e);
        }
    }
​
​
    @Override
    public void close() throws IOException {
        if (connection != null) {
            connection.close();
        }
    }
}

这样处理下层数据，数据不会发生‘’丢失‘’的情况，所以我第一反应是不是缓存的问题导致的，缓存出问题的场景大多数情况是出现线程安全的问题，所以我怀疑是不是线程安全问题，我在程序中添加了线程打印日志，看看是不是多线程处理数据。

System.out.println("---------" + Thread.currentThread().getName() + "---------");

处理数据过程中打印了如下日志：

---------Window(TumblingEventTimeWindows(10000), EventTimeTrigger, BaseOperator, PassThroughWindowFunction) -> (Sink: Print to Std. Out, Process -> (Sink: Unnamed, Sink: Unnamed)) (2/8)---------
---------Window(TumblingEventTimeWindows(10000), EventTimeTrigger, BaseOperator, PassThroughWindowFunction) -> (Sink: Print to Std. Out, Process -> (Sink: Unnamed, Sink: Unnamed)) (4/8)---------
---------Window(TumblingEventTimeWindows(10000), EventTimeTrigger, BaseOperator, PassThroughWindowFunction) -> (Sink: Print to Std. Out, Process -> (Sink: Unnamed, Sink: Unnamed)) (8/8)---------
---------Window(TumblingEventTimeWindows(10000), EventTimeTrigger, BaseOperator, PassThroughWindowFunction) -> (Sink: Print to Std. Out, Process -> (Sink: Unnamed, Sink: Unnamed)) (7/8)---------
---------Window(TumblingEventTimeWindows(10000), EventTimeTrigger, BaseOperator, PassThroughWindowFunction) -> (Sink: Print to Std. Out, Process -> (Sink: Unnamed, Sink: Unnamed)) (1/8)---------
---------Window(TumblingEventTimeWindows(10000), EventTimeTrigger, BaseOperator, PassThroughWindowFunction) -> (Sink: Print to Std. Out, Process -> (Sink: Unnamed, Sink: Unnamed)) (6/8)---------
---------Window(TumblingEventTimeWindows(10000), EventTimeTrigger, BaseOperator, PassThroughWindowFunction) -> (Sink: Print to Std. Out, Process -> (Sink: Unnamed, Sink: Unnamed)) (5/8)---------

可以看出，sink过程中有8个线程做处理，所以此时的我觉得已经找到问题的根因了，代码是多线程执行的，存在线程安全问题，所以我给批量处理的数据做了如下处理，将缓存改成线程安全的集合，同时将字段lastInvokeTime修饰为volatile，至此我以为我已经解决了该问题，但是结果仍然是发送的20条数据只落了部分数据到hbase。

这时我有点穷驴技穷了，我重新review了代码，发现一处代码逻辑漏洞

首先我开始设置批量处理数据时，只有当数据超过500条或者超过时间限制5秒钟我就会将数据都刷入hbase中，问题出现在了超时时间限制。我发送的数据是一次性的发送20条，当不满足500条的时候不会发送，但是触发超时5秒也会，但是这个超时只有在下次消费数据时才会触发这个逻辑，所以我需要定时任务去定时的将缓存中积压的数据刷到hbase，防止source一直没有数据导致缓存中的数据无法下层甚至丢失

知道原因，并且有解决办法就是采用定时任务来，我添加了如下代码：

private final List<Put> puts = Collections.synchronizedList(Lists.newArrayList());
private final  static ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
//定时刷新缓存数据到hbase
public void initFlush() {
        scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {
            try {
                if (CollectionUtils.isNotEmpty(puts) && flag) {
                    Table table = connection.getTable(TableName.valueOf(hbaseStorage.getTable()));
                    table.put(puts);
                    puts.clear();
                    table.close();
                }
            } catch (Exception e) {
                logger.error("刷新缓存错误：", e);
            }
       }, 30, 5, TimeUnit.SECONDS);
 }

问题三：添加定时任务，定时刷缓存，可是数据仍然没有下层到hbase

想法是好的，现实很残酷，运行下来，下层到hbase的数据仍然缺失。我开始怀疑人生了，我debug了代码，发现缓存中的数据为空

如图：

这时候只能去看底层flink的源码了，我通过跟踪堆栈，发现了执行逻辑，sink的invoke方法的执行，在streamSink类中被调用：

@Override
  public void processElement(StreamRecord<IN> element) throws Exception {
    sinkContext.element = element;
    userFunction.invoke(element.getValue(), sinkContext);
  }

我debug到了这边：

如下：

可以看出我上图中用黑色笔圈出的部分，每个线程执行的hbasesink对象都是不一样，也就是说每个线程都有自己的hbasesink对象。问题迎刃而解了，我设置的全局线程池去刷新缓存，是没法针对每个线程中的hbasesink中的缓存操作的，（线程局部变量对其它线程是不可见的）。所以我的做法很简单，我将全局刷新的线程，变为线程局部处理线程，也就是sink线程的子线程，变化逻辑非常简单：

private final List<Put> puts = Collections.synchronizedList(Lists.newArrayList());
​
//标志位
private volatile boolean flag = true;
​
//定时刷新缓存数据到hbase
public void initFlush() {
  final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1);
        scheduledThreadPoolExecutor.scheduleAtFixedRate(() -> {
            try {
                if (CollectionUtils.isNotEmpty(puts) && flag) {
                    Table table = connection.getTable(TableName.valueOf(hbaseStorage.getTable()));
                    table.put(puts);
                    puts.clear();
                    table.close();
                }
            } catch (Exception e) {
                logger.error("刷新缓存错误：", e);
            }
       }, 30, 5, TimeUnit.SECONDS);
 }

如上，我将创建线程的步骤放到了方法里，而非全局变量。

问题四：hbasesink是我自己new的创建的对象，为啥每个sink线程有自己的呢？

这篇关于flink sink 批量下层数据时 ，数据"丢失"？的文章就介绍到这儿，希望我们推荐的文章对大家有所帮助，也希望大家多多支持为之网！