代码之家  ›  专栏  ›  技术社区  ›  krizajb

自定义AllWindowFunction类成员

  •  0
  • krizajb  · 技术社区  · 8 年前

    我有个习惯 AllWindowFunction 具有负责数据库插入的类成员的类。到数据库的连接是持久的,并在构造期间打开。

    问题是 创建/打开连接的实例与在apply事件上调用的实例不同。此解决方案是静态成员,但我想知道这是否是唯一的解决方案?

    示例代码:

    public class CustomWindowFunction implements AllWindowFunction<String, String, TimeWindow> {
    
        private static Connection database;
    
        CustomWindowFunction() {
            database = new Connection();
        }
    
        @Override
        public void apply(TimeWindow timeWindow, Iterable<String> trades, Collector<String> out) {
            // process data
            database.save(data);
            out.collect(data.toString());
        }
    }
    

    我找不到任何关于这个机制的信息,我只知道构造函数中的对象ID和从中调用的对象ID不同 apply .

    1 回复  |  直到 7 年前
        1
  •  2
  •   Dawid Wysakowicz    8 年前

    这是因为每个函数都必须序列化才能分布在集群的节点上。你可以尝试使用 RichAllWindowFunction open() 方法,该方法将在每个并行运算符开始时调用。 在这种方法中,您可以创建连接

    public class CustomWindowFunction implements RichAllWindowFunction<String, String, TimeWindow> {
    
        private Connection database;
    
        @Override
        public void open(Configuration parameters) {
            database = new Connection();
        }
    
        @Override
        public void apply(TimeWindow timeWindow, Iterable<String> trades, Collector<String> out) {
            // process data
            database.save(data);
            out.collect(data.toString());
        }
    }