Here is what i am tring to do.   1. Get list of features store into JavaPairRDD becasue we want featues to be executed in parallel.    final Broadcast <JavaPairRDD<String, String>> broadCastfeatureKeyClassPair = getBroadcastRDD( jsc );     final JavaPairRDD<String, String> featureKeyClassPair =  broadCastfeatureKeyClassPair.value();   2. Get stream of input data from Kafka stream   ....    JavaPairReceiverInputDStream<String, String> messages = KafkaUtils.createStream(jssc, zkQuorum, group, topicMap);   3. For each input stream of data execute list of features that got from the broadcast variable  ...     JavaPairDStream<String, String> urlFeatureScore = lines.mapToPair(       new PairFunction<String, String, String >() {           public Tuple2<String, String> call(final String urlString) throws Exception {             InputJson inputJson=  mapper.readValue(urlString, InputJson.class);             String data = inputJson.getData();             String  featureScore = getFeatureScore(inputJson, featureKeyClassPair);               return new Tuple2<String, String>(data,  featureScore);         }       });       4. Method called in step 3 {This step is failing because RDD is called inside anotehr RDD} private static String  getFeatureScore(InputJson inputJson, JavaPairRDD<String, String> featureKeyClassPair ){                  JavaRDD<Double> featureScoreRDD = featureKeyClassPair.map(                          new Function<Tuple2<String, String>, Double>() {                              public Double call(Tuple2<String, String> keyValue) {                                      Double score= getFeatureScore(keyValue,  urlString,  urlContent);                                  return score;                              }                          });                                        List<Double> list = featureScoreRDD.collect();                   ...     }     return featureScore; }   5. Using reflection to invoke each feature class. private static Double getFeatureScore(Tuple2<String, String> keyValue, String urlString, String urlContent){     String featureKey = (String) keyValue._1();      String className = (String) keyValue._2();      Double score = 0.0;          Class<?> featureClass = Class.forName(className);               Method featureScoreMethod = featureClass.getMethod("getFeatureScore", String.class, String.class);          score = (Double) featureScoreMethod.invoke(featureClass.newInstance(),urlString , urlContent);            return score; }  private static Broadcast <JavaPairRDD<String, String>>  getBroadcastRDD(JavaSparkContext jsc ){     final JavaRDD<String> featurePropertiesRDD = jsc.textFile(FEATURE_PROPERTIES);           JavaPairRDD<String, String> featureKeyClassPair = featurePropertiesRDD.mapToPair(             new PairFunction<String, String, String>() {                 public Tuple2<String, String> call(String property) {                   return new Tuple2(property.split("=")[0], property.split("=")[1]);                 }            });                 Broadcast <JavaPairRDD<String, String>>  broadCastfeatureKeyClassPair = jsc.broadcast(featureKeyClassPair);     return broadCastfeatureKeyClassPair; } ... View more

Thanks so much for your quick response.  I have a stream of data coming from Kafka, for each input data we have to create many features in parallel.   Since we cannot run RDD transformation inside another RDD.  Should we manage our own multithreaded feature execution for each RDD input data.  ... View more