title: Access hdfs file from spark worker node
body: <p>I am working on a spark application which needs to access and update an object stored as a file in hdfs. I'm unable to figure out how can I do it?</p>
  
  <p>If I'm creating a FileSystem hdfs object and using it: </p>
  
  <pre><code>boolean fileExists = hdfs.exists(new org.apache.hadoop.fs.Path(filePath));
  if (fileExists){
      JavaRDD&lt;MyObject&gt; modelRDD = sc.objectFile(filePath);
  }
  </code></pre>
  
  <p>I get:</p>
  
  <pre><code>ERROR Executor: Exception in task 110.0 in stage 1.0 (TID 112)
  </code></pre>
  
  <p>java.lang.NullPointerException</p>
  
  <p>This part of code runs at the worker so I'm assuming it fails because it doesn't have access to the Spark Context. In such a case, how can I access this hdfs file? </p>
  
  <p>this hdfs file resides at the driver node. I can change replace hdfs with hive, storing data as byte array in hive, but even hive context access is not possible from worker node.</p>
  
  <p>Adding full code for better understanding:</p>
  
  <pre><code>public class MyProgram {
      private static JavaSparkContext sc;
      private static HiveContext hiveContext;
      private static String ObjectPersistenceDir = "/metadata/objects";
      private static org.apache.hadoop.fs.FileSystem hdfs;
      private static String NameNodeURI = "hdfs://&lt;mymachineurl&gt;:9000";
  
      // create and maintain a cache of objects for every run session
      //private static HashMap&lt;String, MyObject&gt; cacheObjects;
  
      public static void main(String ... args) {
          System.out.println("Inside constructor: creating Spark context and Hive context");
          System.out.println("Starting Spark context and SQL context");
  
          sc = new JavaSparkContext(new SparkConf());
          hiveContext = new HiveContext(sc);
  
          //cacheObjects= new HashMap&lt;&gt;();
  
          //DataFrame loadedObjects= hiveContext.sql("select id, filepath from saved_objects where name = 'TEST'");
          //List&lt;Row&gt; rows = loadedObjects.collectAsList();
          //for(Row row : rows){
            //  String key =  (String) row.get(0) ;
            //  String value = (String) row.get(1);
            //  JavaRDD&lt;MyObject&gt; objectRDD = sc.objectFile(value);
            //  cacheObjects.put(key, objectRDD.first());
          //}
  
  
          DataFrame partitionedDF  = hiveContext.sql('select * from mydata');
          String partitionColumnName = "id";
          JavaRDD&lt;Row&gt; partitionedRecs = partitionedDF.repartition(partitionedDF.col(partitionColumnName)).javaRDD();
  
  
          FlatMapFunction&lt;Iterator&lt;Row&gt;, MyObject&gt; flatMapSetup = new FlatMapFunction&lt;java.util.Iterator&lt;Row&gt;, MyObject&gt;() {
              List&lt;MyObject&gt; lm_list = new ArrayList&lt;&gt;();
              MyObject object = null;
  
              @Override
              public List&lt;MyObject&gt; call(java.util.Iterator&lt;Row&gt; it) throws Exception {
                  // for every row, create a record and update the object
                  while (it.hasNext()) {
                      Row row = it.next();
  
                      if (object == null) {
                          String objectKey = "" + id;
                          //object = cacheObjects.get(objectKey);
                          String modelPath = ModelPersistenceDir + "/" +'TEST'+ "/" + id;
                          JavaRDD&lt;MyObject&gt; objectRDD = sc.objectFile(objectPath);
                          object = objectRDD.collect().get(0);
  
                          // object not in cache means not already created
                          if(object == null){
                              if (object == null){
                                  ObjectDef objectDef = new ObjectDef('TEST');
                                  object = new MyObject(objectDef);
                              }
                          }
                      }
  
                      /*
                      / some update on object
                      */
  
                      String objectKey = "" + id ;
                      cacheObjects.put(objectKey, object);
  
                      // Algorithm step 2.6 : to save in hive, add to list
                      lm_list.add(object);
                  } // while Has Next ends
                  return lm_list;
              } // Call -- Iterator ends
          };//); //Map Partition Ends
  
          //todo_nidhi put all objects in collectedObject back to hive
          List&lt;MyObject&gt; collectedObject =  partitionedRecs.mapPartitions(flatMapSetup).collect();
  }
  </code></pre>