问题：

Spark Java流-创建Spark Java数据集时出现空指针异常

闻人思聪

2023-03-14

我正在尝试使用JDBC处理Presto上的查询，并将结果集传递回Spark，以便在其上创建临时表。我的结果集在列表中

我从kafka producer获得了json Msg形式的查询。因此，我们在spark中创建了kafka consumer，以获取信息并进行进一步处理。

以下是我的主要功能：

public static void main(String[] args) throws InterruptedException {

    SparkConf conf = new SparkConf();
     conf.setAppName("Wordcount Background");
     conf.setMaster("local");


    //SparkContext sc = SparkContext.getOrCreate(conf);
     SparkSession spark = 
SparkSession.builder().config(conf).getOrCreate();

     JavaSparkContext sc = new 
JavaSparkContext(spark.sparkContext());
     JavaStreamingContext ssc = new JavaStreamingContext(sc, 
Durations.seconds(5));
     SQLContext sqc = new SQLContext(sc);

     Set<String> topics = Collections.singleton("TestTopic");
        Map<String, String> kafkaParams = new HashMap<>();
        kafkaParams.put("metadata.broker.list", "172.20.3.189:9092");

        JavaPairInputDStream<String, String> directKafkaStream = 
KafkaUtils.createDirectStream(ssc,
                String.class, String.class, StringDecoder.class, 
StringDecoder.class, kafkaParams, topics);

        directKafkaStream.foreachRDD(rdd -> {
            //System.out.println("--- New RDD with " + 
rdd.partitions().size()
                  //  + " partitions and " + rdd.count() + " 
records");
            rdd.foreach(record -> {

                SparkkafkaJson sk = new SparkkafkaJson();

                Dataset<String> dfrdd = 
spark.createDataset(sk.process_query(record._2), Encoders.STRING());
                System.out.print(dfrdd);
                //Dataset<Row> df = spark.read().json(dfrdd);
                //df.show();

            });
        });

        ssc.start();
        ssc.awaitTermination();

}

以下是将结果集返回给主函数的process_query方法：

public List<String> process_query(String queryjson) {
    String resstr="";
    String columnValue="";
    List<String> jsonList = new ArrayList<>();
    //List<String> list=new ArrayList<String>();
    try {

        Class.forName(JDBC_DRIVER);
        //Open a connection
        conn = DriverManager.getConnection(DB_URL, USER, PASS);
        //Execute a query
        stmt = conn.createStatement();
        String sql = process_json(queryjson);
        ResultSet res = stmt.executeQuery(sql);
        ResultSetMetaData rsmd = res.getMetaData();
        int columnsNumber = rsmd.getColumnCount();
        //Extract data from result set
        while (res.next()) {
          //System.out.println(res.getString(""));
            Gson userGson=new GsonBuilder().create();
            JsonObject params = new JsonObject();
            for (int i = 1; i <= columnsNumber; i++) {

                String ColName = rsmd.getColumnName(i);
                String ColVal = res.getString(i);


                params.addProperty(ColName, ColVal);

            }

            resstr = userGson.toJson(params);
            jsonList.add(resstr);
        }


        //Clean-up environment
        res.close();
        stmt.close();
        conn.close();
      } catch (SQLException se) {
        //Handle errors for JDBC
        se.printStackTrace();
      } catch (Exception e) {
        //Handle errors for Class.forName
        e.printStackTrace();
      } finally {
        //finally block used to close resources
        try {
          if (stmt != null) stmt.close();
        } catch (SQLException sqlException) {
          sqlException.printStackTrace();
        }
        try {
          if (conn != null) conn.close();
        } catch (SQLException se) {
          se.printStackTrace();
        }
      }
    return jsonList;
}

但我仍然得到了这个错误输出

    2019-05-30 13:17:41 INFO  ContextCleaner:54 - Cleaned accumulator 42
    2019-05-30 13:17:41 INFO  ContextCleaner:54 - Cleaned accumulator 109
    2019-05-30 13:17:43 INFO  CodeGenerator:54 - Code generated in 216.222798 
    ms
    2019-05-30 13:17:43 ERROR Executor:91 - Exception in task 1.0 in stage 9.0 
    (TID 19)
    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.<init>(Dataset.scala:183)
        at org.apache.spark.sql.Dataset$.apply(Dataset.scala:65)
        at org.apache.spark.sql.SparkSession.createDataset(SparkSession.scala:474)
        at org.apache.spark.sql.SparkSession.createDataset(SparkSession.scala:511)
        at SparkkafkaJson.SparkkafkaJson.lambda$1(SparkkafkaJson.java:213)
        at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreach$1.apply(JavaRDDLike.scala:351)
        at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreach$1.apply(JavaRDDLike.scala:351)
        at scala.collection.Iterator$class.foreach(Iterator.scala:891)
        at org.apache.spark.util.NextIterator.foreach(NextIterator.scala:21)
        at org.apache.spark.rdd.RDD$$anonfun$foreach$1$$anonfun$apply$27.apply(RDD.scala:927)
        at org.apache.spark.rdd.RDD$$anonfun$foreach$1$$anonfun$apply$27.apply(RDD.scala:927)
        at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
        at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:2101)
        at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
        at org.apache.spark.scheduler.Task.run(Task.scala:121)
        at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:402)
        at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
        at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:408)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
        at java.lang.Thread.run(Thread.java:748)
2019-05-30 13:17:43 WARN  TaskSetManager:66 - Lost task 1.0 in stage 9.0 (TID 19, localhost, executor driver): 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.<init>(Dataset.scala:183)
        at org.apache.spark.sql.Dataset$.apply(Dataset.scala:65)
        at org.apache.spark.sql.SparkSession.createDataset(SparkSession.scala:474)
        at org.apache.spark.sql.SparkSession.createDataset(SparkSession.scala:511)
        at SparkkafkaJson.SparkkafkaJson.lambda$1(SparkkafkaJson.java:213)
        at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreach$1.apply(JavaRDDLike.scala:351)
        at org.apache.spark.api.java.JavaRDDLike$$anonfun$foreach$1.apply(JavaRDDLike.scala:351)

请帮帮忙

共有1个答案

鱼宜

2023-03-14

通过逻辑方式修复了上述问题，如下所示：

    directKafkaStream.foreachRDD(rdd -> {
            //System.out.println("--- New RDD with " + rdd.partitions().size()
                  //  + " partitions and " + rdd.count() + " records");     
            rdd.collect().forEach(record -> {
                List<String> jsonList2 = new ArrayList<>();
                //System.out.print(sk.process_query(record._2,sk,spark));
                jsonList2 = sk.process_query(record._2,sk,spark);

                 if(jsonList2.size() > 0) {
                     //System.out.print("came here");
                     sk.jsonList3 = jsonList2; 
                     String JsonStr = jsonList2.toString();
                     //System.out.print(JsonStr);
                     System.out.print("Sending Data to API");
                     System.out.print(" ");
                     try {
                        sk.Send_query_data_SCDF(JsonStr);
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                     }else {                 
                         System.out.print("came into else");
                         sk.jsonList3= new ArrayList<String>();
                     }


            });

我试图在RDD的foreach下创建一个数据集，这就是为什么is会给出nullpointer异常，因为数据集的创建需要在驱动端而不是执行端完成。因此，收集RDD结果并将其用于Outside foreach以使其持久化

            Dataset<String> dfrdd = spark.createDataset(sk.jsonList3,           
            Encoders.STRING());

            Dataset<Row> wordsDataFrame = spark.read().json(dfrdd);             

            wordsDataFrame.createOrReplaceTempView("words");


              Dataset<Row> wordCountsDataFrame =
                spark.sql("select * from words limit 10");
              wordCountsDataFrame.show();
              String jsonToReturn = 
              wordCountsDataFrame.toJSON().collectAsList().toString();
              System.out.print(jsonToReturn);


              sk.jsonList3= new ArrayList<String>();
              rdd.unpersist();

        });

类似资料：

呈现genericlistrenderer时出现空指针异常

以下是在sun.reflect.nativeMethodAccessorImpl.Invoke0（本机方法）在sun.reflect.nativeMethodAccessorImpl.Invoke（未知源）在sun.reflect.NativeMethodAccessorImpl.Invoke（未知源）在java.lang.Reflect.Method.Invoke（未知源）在com.codena
在片段中创建Toast后出现空指针异常

我试图展示一个祝酒词，当我的API请求不工作，由于没有互联网连接。首先，我尝试在try块失败时显示一个Toast，然后在catch块中显示这个Toast。我尝试了developer.android.com的方法，但我甚至无法启动应用程序。我在catch-block中的代码是这样的：我似乎对方法有问题，因为它突出显示为红色。这似乎是一个问题，因为我在一个片段内部使用这个。我在使用时读到了一些内容
使用TimePickerDialog时出现空指针异常

> 单击受保护得空onPrepareDialog（int id,Dialog Dialog）受保护的对话框onCreateDialog（int id）如果我对timepickerdialog代码做了什么错误，请告诉我。
Etcd与SparkJava REST API框架集成Java

我试图在SparkJava中使用下面的etcd框架 https://github.com/AdoHe/etcd4j 代码如下所示：但是当我尝试访问下面的url时http://localhost:4567/hello 我得到以下错误我在这里遗漏了什么？etcd在作为独立项目使用main（）函数时工作，但是不与SparkJava一起工作，是否有任何etcd客户端与SparkJava一起工作？
检索现有Mongo DB集合时给出空指针异常

我已经创建了一个带有数据的MongoDB，但是当检索特定集合中的所有文档时，我的程序给出了一个空指针异常。这里是我的代码，userRepository.java 但是在rest api中的TIG是相同的函数，运行良好。这就是我在alert.java中使用UserSedder的方式。
Java：拆箱Integer时出现空指针异常？

问题内容：此代码导致空指针异常。我不知道为什么：我已经在调试器中检查了它，所有局部变量都不为空。怎么会这样呢？BiMap来自Google Collections。问题答案：空指针异常是将的结果拆箱的结果。如果不包含键，则返回“ of type” 。假设分配是给引用的，Java将值拆箱到中，导致空指针异常。您应该检查或用作局部变量类型，以免取消装箱并采取相应措施。适当的机制取决于您的上下文

Spark Java流-创建Spark Java数据集时出现空指针异常

共有1个答案

相关问答

相关文章

相关阅读

相关工具

相关文档