我在spark流作业中使用spark sql在配置单元表中搜索。Kafka流媒体工作正常，没有问题。如果我跑了 hiveContext.runSqlHive(sqlQuery); 外面 directKafkaStream.foreachRDD 它工作正常，没有问题。但我需要在流式处理作业中查找配置单元表。jdbc的使用( jdbc:hive2:// )可以，但我想使用sparksql。
我的源代码的重要位置如下所示：

// set context
SparkConf sparkConf = new SparkConf().setAppName(appName).set("spark.driver.allowMultipleContexts", "true");
SparkContext sparkSqlContext = new SparkContext(sparkConf);
JavaStreamingContext streamingContext = new JavaStreamingContext(sparkConf, Durations.seconds(batchDuration));
HiveContext hiveContext = new HiveContext(sparkSqlContext);

// Initialize Direct Spark Kafka Stream. Starts from top
JavaPairInputDStream<String, String> directKafkaStream =
                KafkaUtils.createDirectStream(streamingContext,
                        String.class,
                        String.class,
                        StringDecoder.class,
                        StringDecoder.class,
                        kafkaParams,
                        topicsSet);

// work on stream                   
directKafkaStream.foreachRDD((Function<JavaPairRDD<String, String>, Void>) rdd -> {
    rdd.foreachPartition(tuple2Iterator -> {
        // get message
        Tuple2<String, String> item = tuple2Iterator.next();

        // lookup
        String sqlQuery = "SELECT something FROM somewhere";
        Seq<String> resultSequence = hiveContext.runSqlHive(sqlQuery);
        List<String> result = scala.collection.JavaConversions.seqAsJavaList(resultSequence);

        });
    return null;
});

// Start the computation
streamingContext.start();
streamingContext.awaitTermination();

我没有得到任何有意义的错误，即使我周围的尝试捕捉。
我希望有人能帮忙-谢谢。
//编辑：解决方案如下所示：

// work on stream                   
directKafkaStream.foreachRDD((Function<JavaPairRDD<String, String>, Void>) rdd -> {
    // driver
    Map<String, String> lookupMap = getResult(hiveContext); //something with hiveContext.runSqlHive(sqlQuery);
    rdd.foreachPartition(tuple2Iterator -> {
        // worker
        while (tuple2Iterator != null && tuple2Iterator.hasNext()) {
            // get message
            Tuple2<String, String> item = tuple2Iterator.next();
            // lookup
            String result = lookupMap.get(item._2());
        }
    });
    return null;
});