createDataFrame из списка <Row>результат выдает исключение NullPointerException - PullRequest
Новая
       6

createDataFrame из списка <Row>результат выдает исключение NullPointerException

0 голосов
/ 02 марта 2020

Я использую spark- sql -2.4.1 версии. У меня есть код что-то вроде ниже. 

    val dataDs = ///have dataset

    val part_dataDs = dataDs.repartition(col("fieldX"));

    StructType schemaType = part_dataDs.schema();

    part_part_dataDs.foreachPartition(itr ->{

    Iterable<Row> rowIt = () -> itr;
    List<Row> objs = StreamSupport.stream(rowIt.spliterator(), false)
                      .collect(Collectors.toList());

    System.out.println("inrow.length: " + objs.size());

    Dataset<Row> partitionData = sparkSession.createDataFrame(objs, schemaType);

    partitionData.show;

}

Ошибка: 

[Executor task launch worker for task 418] ERROR org.apache.spark.executor.Executor - Exception in task 1.0 in stage 21.0 (TID 418)
java.lang.NullPointerException
at org.apache.spark.sql.SparkSession.sessionState$lzycompute(SparkSession.scala:143)
at org.apache.spark.sql.SparkSession.sessionState(SparkSession.scala:141)
at org.apache.spark.sql.Dataset$.ofRows(Dataset.scala:77)
at org.apache.spark.sql.SparkSession.createDataFrame(SparkSession.scala:380)

1 Ответ

1 голос
/ 03 марта 2020

Итак, вы, кажется, неправильно понимаете несколько основных вещей. Две функции: foreachPartition и mapPartition работают с каждым разделом в наборе данных. Переменная itr в вашем 

part_part_dataDs.foreachPartition(itr -> ....

относится к итератору раздела. Вы можете использовать этот итератор для перебора списка строк, как если бы это был список строк.

В принципе, вы могли бы написать что-то вроде этого: 

part_part_dataDs.foreachPartition(itr ->{
  itr.foreach(row -> {
    System.out.println(row.getString(0));
  })
})

, хотя я должен подчеркнуть, что этот фрагмент кода не имеет никакого смысла вообще. Оператор println будет выполняться на каком-то случайном рабочем узле, так что это не то, что вы бы увидели, если бы вы не работали на одном узле. Кроме того, этот пример должен просто использовать foreach вместо foreachPartition, но так как это выглядит как надуманный игрушечный пример, я не могу сказать, действительно ли вам требуется foreachPartition или нет.

...