ze70TQ7k

1/*
2 * Copyright (c) 2009 Concurrent, Inc.
3 *
4 * This work has been released into the public domain
5 * by the copyright holder. This applies worldwide.
6 *
7 * In case this is not legally possible:
8 * The copyright holder grants any entity the right
9 * to use this work for any purpose, without any
10 * conditions, unless such conditions are required by law.
11 */
12
13package cascading.hbase;
14
15import java.io.IOException;
16import java.net.URI;
17import java.net.URISyntaxException;
18
19import cascading.tap.SinkMode;
20import cascading.tap.Tap;
21import cascading.tap.TapException;
22import cascading.tap.hadoop.TapCollector;
23import cascading.tap.hadoop.TapIterator;
24import cascading.tuple.TupleEntryCollector;
25import cascading.tuple.TupleEntryIterator;
26import org.apache.hadoop.fs.Path;
27import org.apache.hadoop.hbase.HBaseConfiguration;
28import org.apache.hadoop.hbase.HColumnDescriptor;
29import org.apache.hadoop.hbase.HTableDescriptor;
30import org.apache.hadoop.hbase.MasterNotRunningException;
31import org.apache.hadoop.hbase.client.HBaseAdmin;
32import org.apache.hadoop.hbase.mapred.TableOutputFormat;
33import org.apache.hadoop.mapred.FileInputFormat;
34import org.apache.hadoop.mapred.JobConf;
35import org.slf4j.Logger;
36import org.slf4j.LoggerFactory;
37
38/**
39 * The HBaseTap class is a {@link Tap} subclass. It is used in conjunction with the {@HBaseFullScheme}
40 * to allow for the reading and writing of data to and from a HBase cluster.
41 */
42public class HBaseTap extends Tap
43  {
44  /** Field LOG */
45  private static final Logger LOG = LoggerFactory.getLogger( HBaseTap.class );
46
47  /** Field SCHEME */
48  public static final String SCHEME = "hbase";
49
50  /** Field hBaseAdmin */
51  private transient HBaseAdmin hBaseAdmin;
52
53  private String tableName;
54
55  /**
56   * Constructor HBaseTap creates a new HBaseTap instance.
57   *
58   * @param tableName       of type String
59   * @param HBaseFullScheme of type HBaseFullScheme
60   */
61  public HBaseTap( String tableName, HBaseScheme HBaseFullScheme )
62    {
63    super(HBaseFullScheme, SinkMode.APPEND);
64    this.tableName = tableName;
65    }
66
67  /**
68   * Constructor HBaseTap creates a new HBaseTap instance.
69   *
70   * @param tableName       of type String
71   * @param HBaseFullScheme of type HBaseFullScheme
72   * @param sinkMode        of type SinkMode
73   */
74  public HBaseTap( String tableName, HBaseScheme HBaseFullScheme, SinkMode sinkMode )
75    {
76    super( HBaseFullScheme, sinkMode );
77    this.tableName = tableName;
78    }
79
80  private URI getURI()
81    {
82    try
83      {
84      return new URI( SCHEME, tableName, null );
85      }
86    catch( URISyntaxException exception )
87      {
88      throw new TapException( "unable to create uri", exception );
89      }
90    }
91
92  public Path getPath()
93    {
94    return new Path( getURI().toString() );
95    }
96
97  public TupleEntryIterator openForRead( JobConf conf ) throws IOException
98    {
99    return new TupleEntryIterator( getSourceFields(), new TapIterator( this, conf ) );
100    }
101
102  public TupleEntryCollector openForWrite( JobConf conf ) throws IOException
103    {
104    return new TapCollector( this, conf );
105    }
106
107  private HBaseAdmin getHBaseAdmin() throws MasterNotRunningException
108    {
109    if( hBaseAdmin == null )
110      hBaseAdmin = new HBaseAdmin( new HBaseConfiguration() );
111
112    return hBaseAdmin;
113    }
114
115  public boolean makeDirs( JobConf conf ) throws IOException
116    {
117    HBaseAdmin hBaseAdmin = getHBaseAdmin();
118
119    // TODO need to add check if the families that are being written to
120    // exists already
121    if( hBaseAdmin.tableExists( tableName ) )
122      return true;
123
124    LOG.debug( "creating hbase table: {}", tableName );
125
126    HTableDescriptor tableDescriptor = new HTableDescriptor( tableName );
127    String columnNames = ( (HBaseScheme) getScheme() ).getColumnNames();
128    
129    String[] familyNames = columnNames.split(" ");
130
131    for( String familyName : familyNames )
132      tableDescriptor.addFamily( new HColumnDescriptor( familyName ) );
133
134    hBaseAdmin.createTable( tableDescriptor );
135
136    return true;
137    }
138
139  public boolean deletePath( JobConf conf ) throws IOException
140    {
141    // eventually keep table meta-data to source table create
142    HBaseAdmin hBaseAdmin = getHBaseAdmin();
143
144    if( !hBaseAdmin.tableExists( tableName ) )
145      return true;
146
147    LOG.debug( "deleting hbase table: {}", tableName );
148
149    hBaseAdmin.disableTable( tableName );
150    hBaseAdmin.deleteTable( tableName );
151
152    return true;
153    }
154
155  public boolean pathExists( JobConf conf ) throws IOException
156    {
157    return getHBaseAdmin().tableExists( tableName );
158    }
159
160  public long getPathModified( JobConf conf ) throws IOException
161    {
162    return System.currentTimeMillis(); // currently unable to find last mod time on a table
163    }
164
165  @Override
166  public void sinkInit( JobConf conf ) throws IOException
167    {
168    LOG.debug( "sinking to table: {}", tableName );
169
170    // do not delete if initialized from within a task
171    if( isReplace() && conf.get( "mapred.task.partition" ) == null )
172      deletePath( conf );
173
174    makeDirs( conf );
175
176    conf.set( TableOutputFormat.OUTPUT_TABLE, tableName );
177    super.sinkInit( conf );
178    }
179
180  @Override
181  public void sourceInit( JobConf conf ) throws IOException
182    {
183    LOG.debug( "sourcing from table: {}", tableName );
184
185    FileInputFormat.addInputPaths( conf, tableName );
186    super.sourceInit( conf );
187    }
188  }