apache-spark - 如何将 Spring 上下文传递给 Spark 工作节点

Question

我对 Spark 很陌生，我有 spring 上下文(使用 ClassPathXmlApplicationContext 将 xml 加载到应用程序上下文中)需要传递给工作节点。我能够创建 Spark 上下文并尝试在我的 flatMap 及其 null 中使用相同的上下文。无论如何我可以将它传递给 Flatmap 函数吗？我正在调用另一个需要这个 spring 上下文的 in spark 框架。

这是一些代码

ApplicationContext context = new ClassPathXmlApplicationContext("spring/rules-engine-spring.xml");

JavaRDD<Row> partitions = r.mapPartitions(
  new FlatMapFunction<Iterator<Row>, Row>() {
      public Iterable<Row> call(Iterator<Row> rowIterator) throws Exception {

          List<Data> factList = new ArrayList<Data>();
          while (rowIterator.hasNext()) {
              Row rw = rowIterator.next();
              Data fact = new Data();
              fact.setGrain(rw.getString(0));
              fact.setRoute(rw.getString(1));
              factList.add(fact);
          }
          Iterable itr = new Iterable() {
                  List l = new ArrayList<Integer>();

                  public Iterator iterator() {
                      return l.iterator();
                  }
              };
          return itr;
      }

  });
List result=partitions.collect();

当我尝试在 FlatMapFunction 中使用上下文时，它的值为 null。在此方法之外，上下文具有值(value)。任何帮助，将不胜感激。

Answer 1

当您使变量瞬变时，它不会被序列化并且在工作人员上不可用。这将解释 flatMap 中的空值。您必须确保序列化程序接收类(并删除 transient )。您也许可以使用 kryo 来序列化该类，即使它不是 Serializable .
通过 http://apache-spark-user-list.1001560.n3.nabble.com/Un-serializable-3rd-party-classes-Spark-Java-td7815.html :

There are a few options:

• Kryo might be able to serialize these objects out of the box, depending what’s inside them. Try turning it on as described at http://spark.apache.org/docs/latest/tuning.html.
• If that doesn’t work, you can create your own “wrapper” objects that implement Serializable, or even a subclass of FlexCompRowMatrix. No need to change the original library.
• If the library has its own serialization functions, you could also use those inside a wrapper object. Take a look at https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/SerializableWritable.scala for an example where we make Hadoop’s Writables serializable.