如何广播 DataFrame？

Question

我使用的是 spark-sql-2.4.1 版本。 创建如下广播变量

Broadcast<Map<String,Dataset>> bcVariable = javaSparkContext.broadcast(//read dataset);

我将 bcVariable 传递给函数

Service.calculateFunction(sparkSession, bcVariable.getValue());


public   static class Service {
        public static calculateFunction(
          SparkSession sparkSession,
          Map<String, Dataset> dataSet ) {

        System.out.println("---> size : " + dataSet.size());  //printing size 1


        for( Entry<String, Dataset> aEntry : dataSet.entrySet() ) {
           System.out.println( aEntry.getKey());   //  printing key 
            aEntry.getValue().show()   // throw null pointer exception
           }
    }

这里有什么问题？如何在函数中传递数据集/数据框？

尝试 2：

Broadcast<Dataset> bcVariable = javaSparkContext.broadcast(//read dataset);

我将 bcVariable 传递给函数

 Service.calculateFunction(sparkSession, bcVariable.getValue());

公共静态类服务{ 公共静态计算函数（ 火花会话火花会话， 数据集数据集）{

    System.out.println("---> size : " + dataSet.size());  // throwing null pointer exception.



}

这里有什么问题？如何在函数中传递数据集/数据框？

尝试 3：

Dataset metaData = //read dataset from oracle table i.e. meta-data.

我将元数据传递给函数

Service.calculateFunction(sparkSession, metaData);

public   static class Service {
        public static calculateFunction(
          SparkSession sparkSession,
          Dataset metaData ) {

        System.out.println("---> size : " + metaData.size());  // throwing null pointer exception.



    }

这里有什么问题？如何在函数中传递数据集/数据框？

Answer 1

要广播的值必须是任何 Scala 对象，但不能是 DataFrame。

Service.calculateFunction(sparkSession, metaData) 在执行器上执行，因此元数据是null（因为它没有被序列化并通过线路从驱动程序发送到执行器）。

广播[T]（值：T）：广播[T]

向集群广播一个只读变量，返回一个org.apache.spark.broadcast.Broadcast对象，以便在分布式函数中读取它。变量只会发送到每个集群一次。

考虑DataFrame 数据抽象来表示以类似 SQL 的语言（数据集 API 或 SQL）描述的分布式计算。除了在驱动程序上可以提交计算以供执行（作为执行程序上的任务）之外，将它放在任何地方都没有任何意义。

您只需使用DataFrame.collect“转换”此计算所代表的数据（以DataFrame 术语表示）。

收集数据后，您可以使用.value 方法进行广播和引用。

---

代码如下：

val dataset = // reading data
Broadcast<Map<String,Dataset>> bcVariable = 
  javaSparkContext.broadcast(dataset.collect);
Service.calculateFunction(sparkSession, bcVariable.getValue());

与您的代码相比，唯一的变化是collect。