Page History

/**
 * An intermediate layer for passing data from the execution engine
 * to the ETL engine.
 */
public abstract class DataReader {
  /**
   * Read data from the execution engine as an object array.
   * @return - array of objects with each column represented as an object
   * @throws Exception
   */
  public abstract Object[] readArrayRecord() throws Exception;
  /**
   * Read data from execution engine as text - as a CSV record.
   * public abstract Object readContent(int type) throws Exception;
   * @return - CSV formatted data.
   * @throws Exception
   */
  public abstract String readTextRecord() throws Exception;
  /**
   * Read data from execution engine as a native format.
   * @return - the content in the native format of the intermediate data
   * format being used.
   * @throws Exception
   */
  public abstract Object readContent() throws Exception;
}

/**
 * An intermediate layer for passing data from the ETL framework
 * to the MR framework.
 */
public abstract class DataWriter {
  /**
   * Write an array of objects into the execution framework
   * @param array - data to be written
   */
  public abstract void writeArrayRecord(Object[] array);
  /**
   * Write data into execution framework as text. The Intermediate Data Format
   * may choose to convert the data to another format based on how the data
   * format is implemented
   * @param text - data represented as CSV text.
   */
  public abstract void writeStringRecord(String text);
  /**
   * Write data in the intermediate data format's native format.
   * @param obj - data to be written
   */
  public abstract void writeRecord(Object obj);
}

 /**
   * Get one row of data.
   *
   * @return - One row of data, represented in the internal/native format of
   *         the intermediate data format implementation.
   */
  public T getData() {
    return data;
  }
  /**
   * Set one row of data. If validate is set to true, the data is validated
   * against the schema.
   *
   * @param obj - A single row of data to be moved.
   */
  public void setData(T obj) {
    this.data = obj;
  }
  /**
   * Get one row of data as CSV text. Use {@link #SqoopIDFUtils} for reading and writing
   * into the sqoop specified CSV text format for each {@link #ColumnType} field in the row
   * Why a "native" internal format and then return CSV text too?
   * Imagine a connector that moves data from a system that stores data as a
   * serialization format called FooFormat. If I also need the data to be
   * written into HDFS as FooFormat, the additional cycles burnt in converting
   * the FooFormat to text and back is useless - so using the sqoop specified
   * CSV text format saves those extra cycles
   * <p/>
   * Most fast access mechanisms, like mysqldump or pgsqldump write the data
   * out as CSV, and most often the source data is also represented as CSV
   * - so having a minimal CSV support is mandated for all IDF, so we can easily read the
   * data out as text and write as text.
   * <p/>
   * @return - String representing the data in CSV text format.
   */
  public abstract String getCSVTextData();
  /**
   * Set one row of data as CSV

public abstract class IntermediateDataFormat<T> {
  protected volatile T data;
  public int hashCode() {
    return data.hashCode();
  }

  /**
   * Get one row of data.
   *
   * @return - One row of data, represented in the internal/native format of
   *         the intermediate data format implementation.
   */
  public T getData() {
    return data;
  }
  /**
   * Set one row of data. If validate is set to true, the data is validated
   * against the schema.
   *
   * @param data - A single row of data to be moved.
   */
  public abstract void setDatasetCSVTextData(TString data) {
    this.data = datacsvText);
  }
  /**
   * Get one row of data as CSVan Object text. Use SqoopDataUtils for reading and writingarray.
   * intoSqoop theuses sqoopdefined specified CSV text format for each {@link #ColumnType} field in the rowobject representation
   * Whyfor aeach "native" internal format and then return CSV text too?column type. For instance org.joda.time to represent date.
   * ImagineUse a{@link connector#SqoopIDFUtils} thatfor movesreading dataand fromwriting ainto systemthe thatsqoop
 stores data as* a
specified object format *for serializationeach format{@link called#ColumnType} FooFormat.field Ifin I also need the data to bethe row
   * </p>
   * written@return into- HDFSString asrepresenting FooFormat, the additionaldata cyclesas burntan inObject convertingarray
   * theIf FooFormatFROM to text and backTO isschema uselessexist, -we sowill usinguse theSchemaMatcher sqoopto specified
get the  * CSV text format saves those extra cyclesdata according to "TO" schema
   * <p/>
  public *abstract Most fast access mechanisms, like mysqldump or pgsqldump write the data
   * out as CSV, and most often the source data is also represented as CSV
   * - so having a minimal CSV support is mandated for all IDF, so we can easily read the
   * data out as text and write as text.
   * <p/>
   * @return - String representing the data in CSV text format.
   Object[] getObjectData();
 /**
  * Set one row of data as an Object array.
  * It also should construct the data representation
  * that the IDF represents so that the object is ready to
  * consume when getData is invoked. Custom implementations
  * will override this method to convert form object array
  * to the data format
  */
  public abstract Stringvoid getCSVTextDatasetObjectData(Object[] data);
  /**
   * Set the oneschema row offor serializing/de-serializing data as CSV.
   *
   */
  public abstract void setCSVTextData(String csvText);
  /**@param schema
   * Get   one row of data as an Object- array.the Sqoopschema usesused defined object representationfor serializing/de-serializing data
   * for each column type. For instance org.joda.time to represent date.Use SqoopDataUtils
   * for reading and writing into the sqoop specified object format
   * for each {@link #ColumnType} field in the row
   * </p>
   * @return - String representing the data as an Object array
   * If FROM and TO schema exist, we will use SchemaMatcher to get the data according to "TO" schema/
  public void setSchema(Schema schema) {
    if (schema == null) {
      // TODO(SQOOP-1956): throw an exception since working without a schema is dangerous
      return;
    }
    this.schema = schema;
  ..
  }
  /**
   * Serialize the fields of this object to <code>out</code>.
   *
   * @param out <code>DataOuput</code> to serialize this object into.
   * @throws IOException
   */
  public abstract Object[]void getObjectDatawrite(DataOutput out) throws IOException;
  /**
   * SetDeserialize onethe rowfields of datathis asobject an Object arrayfrom <code>in</code>.
   *
   */
  public abstract void setObjectData(Object[] data);
  /** <p>For efficiency, implementations should attempt to re-use storage in the
   * Set the schema for serializing/de-serializing  data.existing object where possible.</p>
   *
   * @param schema - the schema used for serializing/de-serializing  datain <code>DataInput</code> to deseriablize this object from.
   * @throws IOException
   */
  public abstract void setSchemaread(SchemaDataInput schemain) throws IOException;
  /**
   * SerializeProvide the fieldsexternal of this object to <code>out</code>.
   *jars that the IDF depends on
   * @param@return outset <code>DataOuput</code> to serialize this object into.of jars
   */
  public *Set<String> @throwsgetJars() IOException{
   */
 return public abstract void write(DataOutput out) throws IOException;
  /**new HashSet<String>();
  }
  @Override
   * Deserialize the fields of this object from <code>in</code>.public int hashCode() {
    final int prime = 31;
   *
 int result *= <p>For1;
 efficiency, implementations should attemptresult to= re-useprime storage* inresult the
+ ((data == *null) existing? object0 where: possible.</p>data.hashCode());
   *
 result = prime * result @param+ in <code>DataInput</code> to deseriablize this object from.((schema == null) ? 0 : schema.hashCode());
   * @throwsreturn IOExceptionresult;
   */
  public abstract void read(DataInput in) throws IOException;
}}