import cascading.tap.TapException; //导入依赖的package包/类
private TupleEntryCollector getCollector(String path) {
  TupleEntryCollector collector = collectors.get(path);

  if (collector != null)
    return collector;

  try {

    collector = createTupleEntrySchemeCollector(flowProcess,
        parent, path);

    flowProcess.increment(Counters.Paths_Opened, 1);
  } catch (IOException exception) {
    throw new TapException("unable to open template path: " + path,
        exception);
  }

  if (collectors.size() > openTapsThreshold)
    purgeCollectors();

  collectors.put(path, collector);

  return collector;
}
 

import cascading.tap.TapException; //导入依赖的package包/类
@Override
public void sourceConfInit(FlowProcess<JobConf> process, JobConf conf) {
  try {
    Path root = getQualifiedPath(conf);
    if (_options.attrs != null && _options.attrs.length > 0) {
      Pail pail = new Pail(_pailRoot);
      for (List<String> attr : _options.attrs) {
        String rel = Utils.join(attr, "/");
        pail.getSubPail(rel); //ensure the path exists
        Path toAdd = new Path(root, rel);
        LOG.info("Adding input path " + toAdd.toString());
        FileInputFormat.addInputPath(conf, toAdd);
      }
    } else {
      FileInputFormat.addInputPath(conf, root);
    }

    getScheme().sourceConfInit(process, this, conf);
    makeLocal(conf, getQualifiedPath(conf), "forcing job to local mode, via source: ");
    TupleSerialization.setSerializations(conf);
  } catch (IOException e) {
    throw new TapException(e);
  }
}
 

import cascading.tap.TapException; //导入依赖的package包/类
@Override
public void sinkConfInit( FlowProcess<JobConf> process, JobConf conf )
{
    if( !isSink() )
        return;

    // do not delete if initialized from within a task
    try {
        if( isReplace() && conf.get( "mapred.task.partition" ) == null && !deleteResource( conf ) )
            throw new TapException( "unable to drop table: " + tableDesc.getTableName() );

        if( !createResource( conf ) )
            throw new TapException( "unable to create table: " + tableDesc.getTableName() );
    } catch(IOException e) {
        throw new TapException( "error while trying to modify table: " + tableDesc.getTableName() );
    }

    if( username == null )
        DBConfiguration.configureDB( conf, driverClassName, connectionUrl );
    else
        DBConfiguration.configureDB( conf, driverClassName, connectionUrl, username, password );

    super.sinkConfInit( process, conf );
}
 

import cascading.tap.TapException; //导入依赖的package包/类
@Override
public boolean createResource( JobConf conf ) throws IOException
{
    if( resourceExists( conf ) )
        return true;

    try
    {
        LOG.info( "creating table: {}", tableDesc.tableName );

        executeUpdate( tableDesc.getCreateTableStatement() );
    }
    catch( TapException exception )
    {
        LOG.warn( "unable to create table: {}", tableDesc.tableName );
        LOG.warn( "sql failure", exception.getCause() );

        return false;
    }

    return resourceExists( conf );
}
 

import cascading.tap.TapException; //导入依赖的package包/类
@Override
public boolean deleteResource( JobConf conf ) throws IOException
{
    if( !isSink() )
        return false;

    if( !resourceExists( conf ) )
        return true;

    try
    {
        LOG.info( "deleting table: {}", tableDesc.tableName );

        executeUpdate( tableDesc.getTableDropStatement() );
    }
    catch( TapException exception )
    {
        LOG.warn( "unable to drop table: {}", tableDesc.tableName );
        LOG.warn( "sql failure", exception.getCause() );

        return false;
    }

    return !resourceExists( conf );
}
 

import cascading.tap.TapException; //导入依赖的package包/类
@Override
public boolean resourceExists( JobConf conf ) throws IOException
{
    if( !isSink() )
        return true;

    try
    {
        LOG.info( "test table exists: {}", tableDesc.tableName );

        executeQuery( tableDesc.getTableExistsQuery(), 0 );
    }
    catch( TapException exception )
    {
        return false;
    }

    return true;
}
 

import cascading.tap.TapException; //导入依赖的package包/类
private void obtainToken(JobConf conf) {
  if (User.isHBaseSecurityEnabled(conf)) {
    String user = conf.getUser();
    LOG.info("obtaining HBase token for: {}", user);
    try {
      UserGroupInformation currentUser = UserGroupInformation.getCurrentUser();
      user = currentUser.getUserName();
      Credentials credentials = conf.getCredentials();
      for (Token t : currentUser.getTokens()) {
        LOG.debug("Token {} is available", t);
        if ("HBASE_AUTH_TOKEN".equalsIgnoreCase(t.getKind().toString()))
          credentials.addToken(t.getKind(), t);
      }
    } catch (IOException e) {
      throw new TapException("Unable to obtain HBase auth token for " + user, e);
    }
  }
}
 

import cascading.tap.TapException; //导入依赖的package包/类
/**
 * Reads the header record from the source file.
 */
@SuppressWarnings("unchecked")
private CSVRecord getHeaderRecord(FlowProcess<JobConf> flowProcess, Tap tap) {
  Tap textLine = new Hfs(new TextLine(new Fields("line")), tap.getFullIdentifier(flowProcess.getConfigCopy()));

  try (TupleEntryIterator iterator = textLine.openForRead(flowProcess)) {
    String line = iterator.next().getTuple().getString(0);
    boolean skipHeaderRecord = format.getSkipHeaderRecord();
    CSVRecord headerRecord = CSVParser.parse(line, format.withSkipHeaderRecord(false)).iterator().next();
    format.withSkipHeaderRecord(skipHeaderRecord);
    return headerRecord;
  } catch (IOException e) {
    throw new TapException(e);
  }
}
 

import cascading.tap.TapException; //导入依赖的package包/类
@SuppressWarnings("unchecked")
@Override
public boolean source(final FlowProcess<JobConf> flowProcess, final SourceCall<Object[], RecordReader> sourceCall) throws IOException {
    final Object[] context = sourceCall.getContext();

    while (sourceCall.getInput().next(context[0], context[1])) {
        if (hasHeader && ((LongWritable) context[0]).get() == 0)
            continue;
        final String[] split;
        try {
            split = ((CSVParser) sourceCall.getContext()[3]).parseLine(makeEncodedString(context));
        } catch (IOException exc) {
            if (strict)
                throw exc;
            LOGGER.warn("exception", exc);
            flowProcess.increment("com.tresata.cascading.scheme.OpenCsvScheme", "Invalid Records", 1);
            continue;
        }
        for (int i = 0; i < split.length; i++)
            if (split[i].equals(""))
                split[i] = null;
        final Tuple tuple = sourceCall.getIncomingEntry().getTuple();
        tuple.clear();
        tuple.addAll(split);
        if (tuple.size() != getSourceFields().size()) {
            if (strict)
                throw new TapException(String.format("expected %s items but got %s", getSourceFields().size(), tuple.size()), tuple);
            LOGGER.warn("expected {} items but got {}, tuple: {}", new Object[]{getSourceFields().size(), tuple.size() , tuple});
            flowProcess.increment("com.tresata.cascading.scheme.OpenCsvScheme", "Invalid Records", 1);
            tuple.clear(); // probably not necessary
            continue;
        }
        return true;
    }
    return false;
}
 

import cascading.tap.TapException; //导入依赖的package包/类
@Override public void sinkConfInit(FlowProcess<JobConf> flowProcess,
    Tap<JobConf, RecordReader, OutputCollector> tap, JobConf conf) {
  conf.setOutputFormat(PailOutputFormat.class);
  Utils.setObject(conf, PailOutputFormat.SPEC_ARG, getSpec());
  try {
    Pail.create(getFileSystem(conf), _pailRoot, getSpec(), _options.failOnExists);
  } catch (IOException e) {
    throw new TapException(e);
  }
}
 

import cascading.tap.TapException; //导入依赖的package包/类
@Override
public void close() {
    try {
        LOG.info( "closing tap collector for: {}", tap );
        writer.close( reporter );
    } catch( IOException exception ) {
        LOG.warn( "exception closing: {}", exception );
        throw new TapException( "exception closing JDBCTapCollector", exception );
    } finally {
        super.close();
    }
}
 

import cascading.tap.TapException; //导入依赖的package包/类
@Override
public TupleEntryCollector openForWrite( FlowProcess<JobConf> flowProcess, OutputCollector output ) throws IOException {
    if( !isSink() )
        throw new TapException( "this tap may not be used as a sink, no TableDesc defined" );

    LOG.info("Creating JDBCTapCollector output instance");
    JDBCTapCollector jdbcCollector = new JDBCTapCollector( flowProcess, this );

    jdbcCollector.prepare();

    return jdbcCollector;
}
 

import cascading.tap.TapException; //导入依赖的package包/类
@Override
public void sinkConfInit( FlowProcess<JobConf> process, Tap<JobConf, RecordReader, OutputCollector> tap,
    JobConf conf ) {
    if( selectQuery != null )
        throw new TapException( "cannot sink to this Scheme" );

    String tableName = ( (JDBCTap) tap ).getTableName();
    int batchSize = ( (JDBCTap) tap ).getBatchSize();
    DBOutputFormat.setOutput( conf, DBOutputFormat.class, tableName, columns, updateBy, batchSize );

    if( outputFormatClass != null )
        conf.setOutputFormat( outputFormatClass );
}
 

import cascading.tap.TapException; //导入依赖的package包/类
@Override
public void close() {
  try {
    LOG.info("closing tap collector for: {}", tap);
    writer.close(reporter);
  } catch (IOException exception) {
    LOG.warn("exception closing: {}", exception);
    throw new TapException("exception closing HBaseTapCollector", exception);
  } finally {
    super.close();
  }
}
 

import cascading.tap.TapException; //导入依赖的package包/类
@Override
public void sinkConfInit(FlowProcess<JobConf> prcs,
    Tap<JobConf, RecordReader, OutputCollector> tap, JobConf conf) {
  throw new TapException("cannot use as a sink");
}
 

import cascading.tap.TapException; //导入依赖的package包/类
@Override
public void sink(FlowProcess<JobConf> prcs, SinkCall<Object[], OutputCollector> sc)
    throws IOException {
  throw new TapException("cannot use as a sink");
}