title: SparkSession null point exception in Dataset foreach
body: <p>I am new to Spark.</p>
  <p>I want to keep getting message from kafka, and then save to S3 once the message size over 100000.</p>
  <p>I implemented it by Dataset.collectAsList(), but it throw error with <code>Total size of serialized results of 3 tasks (1389.3 MiB) is bigger than spark.driver.maxResultSize</code></p>
  <p>So I turned to use foreach, and it said null point exception when used SparkSession to createDataFrame.</p>
  <p>Any idea about it? Thanks.</p>
  <p>---Code---</p>
  <pre><code>SparkSession spark = generateSparkSession();
          registerUdf4AddPartition(spark);
          Dataset&lt;Row&gt; dataset = spark.readStream().format(&quot;kafka&quot;)
                  .option(&quot;kafka.bootstrap.servers&quot;, args[0])
                  .option(&quot;kafka.group.id&quot;, args[1])
                  .option(&quot;subscribe&quot;, args[2])
                  .option(&quot;kafka.security.protocol&quot;, SecurityProtocol.SASL_PLAINTEXT.name)
                  .load();
          DataStreamWriter&lt;Row&gt; console = dataset.toDF().writeStream().foreachBatch((rawDataset, time) -&gt; {
              Dataset&lt;Row&gt; rowDataset = rawDataset.selectExpr(&quot;CAST(value AS STRING)&quot;);
              //using foreach
              rowDataset.foreach(row -&gt; {
                  List&lt;Span&gt; rawDataList = new CsvToBeanBuilder(new StringReader(row.getString(0))).withType(Span.class).build().parse();
                  spans.addAll(rawDataList);
                  batchSave(spark);
              });
  
              // using collectAsList
              List&lt;Row&gt; rows = rowDataset.collectAsList();
              for (Row row : rows) {
                  List&lt;Span&gt; rawDataList = new CsvToBeanBuilder(new StringReader(row.getString(0))).withType(Span.class).build().parse();
                  spans.addAll(rawDataList);
                  batchSave(spark);
              }
          });
          StreamingQuery start = console.start();
          start.awaitTermination();
  
  public static void batchSave(SparkSession spark){
          synchronized (spans){
              if(spans.size() == 100000){
                  System.out.println(spans.isEmpty());
                  Dataset&lt;Row&gt; spanDataSet = spark.createDataFrame(spans, Span.class);
                  Dataset&lt;Row&gt; finalResult = addCustomizedTimeByUdf(spanDataSet);
  
                  StringBuilder pathBuilder = new StringBuilder(&quot;s3a://fwk-dataplatform-np/datalake/log/FWK/ART2/test/leftAndRight&quot;);
                  finalResult.repartition(1).write().partitionBy(&quot;year&quot;,&quot;month&quot;,&quot;day&quot;,&quot;hour&quot;).format(&quot;csv&quot;).mode(&quot;append&quot;).save(pathBuilder.toString());
                  spans.clear();
              }
          }
      }
  </code></pre>