title: How to use spark Java API to read binary file stream from HDFS?
body: <p>I am writing a component which needs to get the new binary file in a specific HDFS path, so that I can do some online learning based on this data. So, I want to read binary file created by Flume from HDFS in stream. I found several functions provided by spark API, such as </p>
  
  <pre><code>public JavaDStream&lt;byte[]&gt; binaryRecordsStream(String directory,int recordLength)
  </code></pre>
  
  <p>and </p>
  
  <pre><code>public &lt;K,V,F extends org.apache.hadoop.mapreduce.InputFormat&lt;K,V&gt;&gt; 
       JavaPairInputDStream&lt;K,V&gt; fileStream(String directory, Class&lt;K&gt; kClass, Class&lt;V&gt; vClass, Class&lt;F&gt; fClass)
  </code></pre>
  
  <p>But, I really do not know how to use these functions. I have tried <code>binaryRecordStream</code>, but it defines the specific length of file, so it is not good.</p>
  
  <p>For <code>fileStream</code> function, I have used:</p>
  
  <pre><code>SparkConf sparkConf = new   SparkConf().setAppName("SparkFileStreamTest").setMaster("local[2]");
  
  // Create the context with the specified batch size
  JavaStreamingContext jssc = new JavaStreamingContext(sparkConf, new Duration(durationInMillis));
  
  JavaPairInputDStream&lt;LongWritable, BytesWritable&gt; inputDStream = jssc.fileStream(hdfsPath, LongWritable.class, BytesWritable.class, CustomInputFormat.class);
  
  //**********************************************************************
  JavaPairInputDStream&lt;LongWritable, BytesWritable&gt; inputDStream = jssc.fileStream(
              hdfsPath, LongWritable.class, BytesWritable.class, CustomInputFormat.class);
  
  JavaDStream&lt;byte[]&gt; content = inputDStream.map(new Function&lt;Tuple2&lt;LongWritable, BytesWritable&gt;, byte[]&gt;() {
      @Override
      public byte[] call(Tuple2&lt;LongWritable, BytesWritable&gt; tuple2) {
          System.out.println("----------------[testReadFileStreamFromHDFS] ENTER ......");
          if (tuple2 == null) {
              System.out.println("----------------[testReadFileStreamFromHDFS] TUPLE = NULL");
              System.out.println("----------------[testReadFileStreamFromHDFS] END.");
              return null;
          }
          else {
              System.out.println("----------------[testReadFileStreamFromHDFS] KEY = [" + tuple2._1().toString() + "]");
              System.out.println("----------------[testReadFileStreamFromHDFS] VAL-LENGTH = [" + tuple2._2().getBytes().length + "]");
              System.out.println("----------------[testReadFileStreamFromHDFS] END.");
              return tuple2._2().getBytes();
          }
      }
  });
  
  /***********************************************************************/
  if (content == null) {
          System.out.println("----------------[testReadFileStreamFromHDFS] CONTENT = NULL");
  }
  else {
      System.out.println("----------------[testReadFileStreamFromHDFS] CONTENT-length = [" + content.count());
      content.print();
  }
  
  System.out.println("----------------[testReadFileStreamFromHDFS] END-111.");
  
  jssc.start();
  jssc.awaitTermination();
  System.out.println("----------------[testReadFileStreamFromHDFS] END-222.");
  </code></pre>
  
  <p>For <code>CustomInputFormat</code>, I created</p>
  
  <pre><code>public class CustomInputFormat extends FileInputFormat&lt;LongWritable, BytesWritable&gt; {
  
  private CustomInputSplit mInputSplit;
  
  public CustomInputFormat() {
      mInputSplit = new CustomInputSplit();
  }
  
  @Override
  public List&lt;InputSplit&gt; getSplits(JobContext context)
          throws IOException {
  
      System.out.println("----------------[CustomInputFormat] 1111 ......");
      final ArrayList&lt;InputSplit&gt; result = new ArrayList&lt;InputSplit&gt;();
      result.add(mInputSplit);
  
      System.out.println("----------------[CustomInputFormat] 2222 ......");
      return result;
  }
  
  @Override
  public RecordReader&lt;LongWritable, BytesWritable&gt; createRecordReader(
          InputSplit inputSplit, TaskAttemptContext taskAttemptContext)
          throws IOException, InterruptedException {
  
      System.out.println("----------------[CustomInputFormat] 3333 ......");
      System.out.println("----------------[CustomInputFormat] ENTER createRecordReader, inputSplit-length = ["
              + inputSplit.getLength() + "]");
  
      mInputSplit.init(inputSplit);
  
      System.out.println("----------------[CustomInputFormat] 4444 ......");
      return new CustomRecordReader();
  }
  
  @Override
  protected boolean isSplitable(JobContext context, Path filename) {
      System.out.println("----------------[CustomInputFormat] 5555 ......");
      return false;
  }
  
  public class CustomRecordReader extends RecordReader&lt;LongWritable, BytesWritable&gt; {
  
  private BytesWritable mValues;
  private int mCursor;
  
  public CustomRecordReader() {
      System.out.println("----------------[CustomRecordReader] 1111 ......");
      mValues = null;
      mCursor = 0;
      System.out.println("----------------[CustomRecordReader] 2222 ......");
  }
  
  @Override
  public void initialize(InputSplit inputSplit, TaskAttemptContext taskAttemptContext)
          throws IOException, InterruptedException {
      System.out.println("----------------[CustomRecordReader] 3333 ......");
      CustomInputSplit customInputSplit = (CustomInputSplit) inputSplit;
      mValues = customInputSplit.getValues();
      System.out.println("----------------[CustomRecordReader] 4444 ......");
  }
  
  @Override
  public boolean nextKeyValue() throws IOException, InterruptedException {
      System.out.println("----------------[CustomRecordReader] 5555 ......");
      boolean existNext = (mCursor == 0);
      mCursor++;
      System.out.println("----------------[CustomRecordReader] 6666 ......");
      return existNext;
  }
  
  @Override
  public LongWritable getCurrentKey() throws IOException, InterruptedException {
      System.out.println("----------------[CustomRecordReader] 7777 ......");
      return new LongWritable(0);
  }
  
  @Override
  public BytesWritable getCurrentValue() throws IOException, InterruptedException {
      System.out.println("----------------[CustomRecordReader] 8888 ......");
      return mValues;
  }
  
  @Override
  public float getProgress() throws IOException, InterruptedException {
      System.out.println("----------------[CustomRecordReader] 9999 ......");
      return 0;
  }
  
  @Override
  public void close() throws IOException {
      System.out.println("----------------[CustomRecordReader] AAAA ......");
      mValues = null;
  }
  }
  
  public class CustomInputSplit extends InputSplit implements Writable {
  
  private long mLength;
  private String[] mLocations;
  
  private final BytesWritable mContent;
  
  public CustomInputSplit() {
      System.out.println("----------------[CustomInputSplit] 1111 ......");
      mLength = 0;
      mLocations = null;
  
      mContent = new BytesWritable();
      System.out.println("----------------[CustomInputSplit] 2222 ......");
  }
  
  public void init(InputSplit inputSplit) throws IOException, InterruptedException {
      System.out.println("----------------[CustomInputSplit] 3333 ......");
      mLength = inputSplit.getLength();
  
      String[] locations = inputSplit.getLocations();
      if (locations != null) {
          int numLocations = locations.length;
          mLocations = new String[numLocations];
          for (int i = 0; i &lt; numLocations; i++) {
              mLocations[i] = locations[i];
          }
      }
      System.out.println("----------------[CustomInputSplit] 4444 ......");
  }
  
  @Override
  public long getLength() throws IOException, InterruptedException {
      System.out.println("----------------[CustomInputSplit] 5555 ......");
      return mLength;
  }
  
  @Override
  public String[] getLocations() throws IOException, InterruptedException {
      if (mLocations == null) {
          System.out.println("----------------[CustomInputSplit] 6666-0001 ...... mLocations = [NULL]");
          mLocations = new String[] {"localhost"};
      }
      System.out.println("----------------[CustomInputSplit] 6666-0002 ...... mLocations-length = [" + mLocations.length + "]");
      return mLocations;
  }
  
  @Override
  public void write(DataOutput dataOutput) throws IOException {
      System.out.println("----------------[CustomInputSplit] 7777 ......");
      mContent.write(dataOutput);
  }
  
  @Override
  public void readFields(DataInput dataInput) throws IOException {
      System.out.println("----------------[CustomInputSplit] 8888 ......");
      mContent.readFields(dataInput);
  }
  
  public BytesWritable getValues() {
      System.out.println("----------------[CustomInputSplit] 9999 ......");
      return mContent;
  }
  }
  </code></pre>
  
  <p>But when I print:</p>
  
  <p><code>System.out.println("----------------[testReadFileStreamFromHDFS] VAL-LENGTH = [" + tuple2._2().getBytes().length + "]");</code></p>
  
  <p>I always get 0 length:</p>
  
  <p><code>----------------[testReadFileStreamFromHDFS] VAL-LENGTH = [0]</code></p>
  
  <p>Are there some problems with <code>CustomerInputFormat.class</code>? Does anybody know how to use Spark stream Java API to read binary file from HDFS?</p>