java/pig-galago/src/com/yahoo/pig/builtin/StorageText.java

   1 /*
   2  * Copyright (c) 2007 Yahoo! Inc.  All rights reserved.
   3  * See accompanying LICENSE file.
   4  */
   5 package com.yahoo.pig.builtin;
   6
   7 import java.io.BufferedReader;
   8 import java.io.IOException;
   9 import java.io.InputStream;
  10 import java.io.InputStreamReader;
  11 import java.io.OutputStream;
  12
  13 import com.yahoo.pig.StorageFunc;
  14 import com.yahoo.pig.PigServer;
  15 import com.yahoo.pig.data.DataAtom;
  16 import com.yahoo.pig.data.Tuple;
  17 import com.yahoo.pig.impl.io.FileLocalizer;
  18 import com.yahoo.pig.impl.io.InputStreamPosition;
  19
  20 /**
  21  * This load function simply creates a tuple for each line of text that has a single field that
  22  * contains the line of text.
  23  */
  24 public class StorageText extends StorageFunc {
  25     BufferedReader      fr;
  26     InputStream         fsis;
  27     InputStreamPosition posInputStream;
  28     long                end;
  29
  30     public void bindTo(InputStream is, long offset, long end) throws IOException {
  31         posInputStream = new InputStreamPosition(is, offset);
  32         fr = new BufferedReader(new InputStreamReader(posInputStream));
  33         // Since we are not block aligned we throw away the first
  34         // record and cound on a different instance to read it
  35         if (offset != 0)
  36             getNext();
  37         this.end = end;
  38         this.fsis = is;
  39     }
  40
  41     public Tuple getNext() throws IOException {
  42         if (fsis != null && posInputStream.getPosition() > end)
  43             return null;
  44         String line;
  45         if ((line = fr.readLine()) != null) {
  46             Tuple t = new Tuple(1);
  47             t.setField(0, new DataAtom(line));
  48             return t;
  49         }
  50
  51         return null;
  52     }
  53
  54     OutputStream os;
  55         @Override
  56         public void bindTo(OutputStream os) throws IOException {
  57                 this.os = os;
  58         }
  59
  60         @Override
  61         public void done() throws IOException {
  62         }
  63
  64         @Override
  65         public void putNext(Tuple f) throws IOException {
  66                 os.write(f.toDelimitedString(", ").getBytes());
  67         }
  68 }