java/pig-galago/src/com/yahoo/pig/impl/io/PigFile.java

   1 /*
   2  * Copyright (c) 2007 Yahoo! Inc.  All rights reserved.
   3  * See accompanying LICENSE file.
   4  */
   5 package com.yahoo.pig.impl.io;
   6
   7 import java.io.BufferedOutputStream;
   8 import java.io.IOException;
   9 import java.io.InputStream;
  10 import java.util.Iterator;
  11
  12 import com.yahoo.pig.StorageFunc;
  13 import com.yahoo.pig.builtin.PigStorage;
  14 import com.yahoo.pig.data.BagFactory;
  15 import com.yahoo.pig.data.DataBag;
  16 import com.yahoo.pig.data.Tuple;
  17 import com.yahoo.pig.PigServer.ExecType;
  18
  19 public class PigFile {
  20     private String file = null;
  21     private String delimiterExpression = null;
  22     boolean append = false;
  23
  24     public PigFile(String filename, boolean append) {
  25         file = filename;
  26         this.append = append;
  27     }
  28
  29     public PigFile(String filename){
  30         file = filename;
  31     }
  32
  33     public PigFile(String filename, String delimiterExpression) {
  34         file = filename;
  35         this.delimiterExpression = delimiterExpression;
  36     }
  37
  38     public DataBag load(ExecType execType) throws IOException {
  39
  40         StorageFunc lfunc = delimiterExpression == null ? new PigStorage() : new PigStorage(delimiterExpression);
  41         return load(execType, lfunc);
  42     }
  43
  44     public DataBag load(ExecType execType, StorageFunc lfunc) throws IOException {
  45         DataBag content = BagFactory.getInstance().getNewBag();
  46         InputStream is = FileLocalizer.open(execType, file);
  47         lfunc.bindTo(is, 0, Long.MAX_VALUE);
  48         Tuple f = null;
  49         while ((f = lfunc.getNext()) != null) {
  50             content.add(f);
  51         }
  52         return content;
  53     }
  54
  55     public void store(ExecType execType, DataBag data) throws IOException {
  56         store(execType, data, new PigStorage());
  57     }
  58
  59     public void store(ExecType execType, DataBag data, StorageFunc sfunc) throws IOException {
  60         BufferedOutputStream bos = new BufferedOutputStream(FileLocalizer.create(execType, file, append));
  61         sfunc.bindTo(bos);
  62         for (Iterator<Tuple> it = data.content(); it.hasNext();) {
  63             Tuple row = it.next();
  64             sfunc.putNext(row);
  65         }
  66         sfunc.done();
  67         bos.close();
  68     }
  69
  70 }