import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
private HCatSchema generateHCatTableSchema(ColumnGenerator... extraCols)
  throws Exception {
  List<HCatFieldSchema> hCatTblCols = new ArrayList<HCatFieldSchema>();
  hCatTblCols.clear();
  hCatTblCols.add(new HCatFieldSchema("id", HCatFieldSchema.Type.INT, ""));
  hCatTblCols
    .add(new HCatFieldSchema("msg", HCatFieldSchema.Type.STRING, ""));
  for (ColumnGenerator gen : extraCols) {
    if (gen.getKeyType() == KeyType.NOT_A_KEY) {
      hCatTblCols
        .add(new HCatFieldSchema(gen.getName(), gen.getHCatType(), ""));
    }
  }
  HCatSchema hCatTblSchema = new HCatSchema(hCatTblCols);
  return hCatTblSchema;
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
private HCatSchema generateHCatDynamicPartitionSchema(
  ColumnGenerator... extraCols) throws Exception {
  List<HCatFieldSchema> hCatPartCols = new ArrayList<HCatFieldSchema>();
  hCatPartCols.clear();
  boolean staticFound = false;
  for (ColumnGenerator gen : extraCols) {
    if (gen.getKeyType() != KeyType.NOT_A_KEY) {
      if (gen.getKeyType() == KeyType.STATIC_KEY && !staticFound) {
        staticFound = true;
        continue;
      }
      hCatPartCols
        .add(new HCatFieldSchema(gen.getName(), gen.getHCatType(), ""));
    }
  }
  HCatSchema hCatPartSchema = new HCatSchema(hCatPartCols);
  return hCatPartSchema;

}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
private List<HCatRecord> generateHCatRecords(int numRecords,
  HCatSchema hCatTblSchema, ColumnGenerator... extraCols) throws Exception {
  List<HCatRecord> records = new ArrayList<HCatRecord>();
  List<HCatFieldSchema> hCatTblCols = hCatTblSchema.getFields();
  int size = hCatTblCols.size();
  for (int i = 0; i < numRecords; ++i) {
    DefaultHCatRecord record = new DefaultHCatRecord(size);
    record.set(hCatTblCols.get(0).getName(), hCatTblSchema, i);
    record.set(hCatTblCols.get(1).getName(), hCatTblSchema, "textfield" + i);
    boolean staticFound = false;
    int idx = 0;
    for (int j = 0; j < extraCols.length; ++j) {
      if (extraCols[j].getKeyType() == KeyType.STATIC_KEY
        && !staticFound) {
        staticFound = true;
        continue;
      }
      record.set(hCatTblCols.get(idx + 2).getName(), hCatTblSchema,
        extraCols[j].getHCatValue(i));
      ++idx;
    }

    records.add(record);
  }
  return records;
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
/**
 * Create an {@link org.apache.hcatalog.mapreduce.HCatRecordReader}.
 *
 * @param split Input split
 * @param schema Table schema
 * @param taskContext Context
 * @return Record reader
 * @throws IOException
 * @throws InterruptedException
 */
private RecordReader<WritableComparable, HCatRecord>
createRecordReader(InputSplit split,
                   HCatSchema schema,
                   TaskAttemptContext taskContext)
  throws IOException, InterruptedException {
  HCatSplit hcatSplit = HCatUtils.castToHCatSplit(split);
  PartInfo partitionInfo = hcatSplit.getPartitionInfo();
  JobContext jobContext = taskContext;
  Configuration conf = jobContext.getConfiguration();

  HCatStorageHandler storageHandler = HCatUtil.getStorageHandler(
      conf, partitionInfo);

  JobConf jobConf = HCatUtil.getJobConfFromContext(jobContext);
  Map<String, String> jobProperties = partitionInfo.getJobProperties();
  HCatUtil.copyJobPropertiesToJobConf(jobProperties, jobConf);

  Map<String, String> valuesNotInDataCols = getColValsNotInDataColumns(
      schema, partitionInfo);

  return HCatUtils.newHCatReader(storageHandler, valuesNotInDataCols);
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
/**
 * Get values for fields requested by output schema which will not be in the
 * data.
 *
 * @param outputSchema Output schema
 * @param partInfo Partition info
 * @return Values not in data columns
 */
private static Map<String, String> getColValsNotInDataColumns(
    HCatSchema outputSchema,
    PartInfo partInfo) {
  HCatSchema dataSchema = partInfo.getPartitionSchema();
  Map<String, String> vals = new HashMap<String, String>();
  for (String fieldName : outputSchema.getFieldNames()) {
    if (dataSchema.getPosition(fieldName) == null) {
      // this entry of output is not present in the output schema
      // so, we first check the table schema to see if it is a part col
      if (partInfo.getPartitionValues().containsKey(fieldName)) {
        vals.put(fieldName, partInfo.getPartitionValues().get(fieldName));
      } else {
        vals.put(fieldName, null);
      }
    }
  }
  return vals;
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
/**
 * Extract partition info.
 *
 * @param schema Table schema
 * @param sd Storage descriptor
 * @param parameters Parameters
 * @param conf Configuration
 * @param inputJobInfo Input job info
 * @return Partition info
 * @throws IOException
 */
private static PartInfo extractPartInfo(
    HCatSchema schema, StorageDescriptor sd, Map<String, String> parameters,
    Configuration conf, InputJobInfo inputJobInfo) throws IOException {
  StorerInfo storerInfo = InternalUtil.extractStorerInfo(sd, parameters);

  Properties hcatProperties = new Properties();
  HCatStorageHandler storageHandler = HCatUtil.getStorageHandler(conf,
      storerInfo);

  // Copy the properties from storageHandler to jobProperties
  Map<String, String> jobProperties =
      HCatUtil.getInputJobProperties(storageHandler, inputJobInfo);

  for (Map.Entry<String, String> param : parameters.entrySet()) {
    hcatProperties.put(param.getKey(), param.getValue());
  }

  return new PartInfo(schema, storageHandler, sd.getLocation(),
      hcatProperties, jobProperties, inputJobInfo.getTableInfo());
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
/**
 * Ensure we can do a basic serialization of a STRUCT
 *
 * @throws Exception
 */
@Test
public void testSerializeStruct() throws Exception {
    HCatTable table = mock(HCatTable.class);

    HCatSchema structSchema = new HCatSchema(Lists.newArrayList(
            getSubFieldSchema("intField", HCatFieldSchema.Type.BIGINT),
            getSubFieldSchema("strField", HCatFieldSchema.Type.STRING)));
    HCatFieldSchema fieldSchema = new HCatFieldSchema("testField", HCatFieldSchema.Type.STRUCT, structSchema, "");

    HiveSerializer serializer = new HiveSerializer(table);

    Map<String, Object> testMap = Maps.newHashMap();
    testMap.put("intField", 123L);
    testMap.put("strField", "This is a string");

    byte[] result = serializer.serializeStruct(fieldSchema, testMap, 1);

    ByteArrayOutputStream expectedResult = new ByteArrayOutputStream();
    expectedResult.write(Bytes.toBytes(123L));
    expectedResult.write('\002');
    expectedResult.write(Bytes.toBytes("This is a string"));

    Assert.assertArrayEquals(expectedResult.toByteArray(), result);
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
/**
 * Ensure we throw an exception when a field within a STRUCT is null, as this is not allowed.
 *
 * @throws Exception
 */
@Test(expected=IllegalArgumentException.class)
public void testSerializeStructWithNullField() throws Exception {
    HCatTable table = mock(HCatTable.class);

    HCatSchema structSchema = new HCatSchema(Lists.newArrayList(
            getSubFieldSchema("intField", HCatFieldSchema.Type.BIGINT),
            getSubFieldSchema("strField", HCatFieldSchema.Type.STRING)));
    HCatFieldSchema fieldSchema = new HCatFieldSchema("testField", HCatFieldSchema.Type.STRUCT, structSchema, "");

    HiveSerializer serializer = new HiveSerializer(table);

    Map<String, Object> testMap = Maps.newHashMap();

    serializer.serializeStruct(fieldSchema, testMap, 1);
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
private void loadHCatTable(HCatSchema hCatSchema, String table,
  int count, ColumnGenerator... extraCols)
  throws Exception {
  Map<String, String> staticKeyMap = new HashMap<String, String>();
  for (ColumnGenerator col : extraCols) {
    if (col.getKeyType() == KeyType.STATIC_KEY) {
      staticKeyMap.put(col.getName(), (String) col.getHCatValue(0));
    }
  }
  loadHCatTable(null, table, staticKeyMap,
    hCatSchema, generateHCatRecords(count, hCatSchema, extraCols));
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
private HCatSchema generateHCatPartitionSchema(ColumnGenerator... extraCols)
  throws Exception {
  List<HCatFieldSchema> hCatPartCols = new ArrayList<HCatFieldSchema>();

  for (ColumnGenerator gen : extraCols) {
    if (gen.getKeyType() != KeyType.NOT_A_KEY) {
      hCatPartCols
        .add(new HCatFieldSchema(gen.getName(), gen.getHCatType(), ""));
    }
  }
  HCatSchema hCatPartSchema = new HCatSchema(hCatPartCols);
  return hCatPartSchema;
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
private HCatSchema generateHCatStaticPartitionSchema(
  ColumnGenerator... extraCols) throws Exception {
  List<HCatFieldSchema> hCatPartCols = new ArrayList<HCatFieldSchema>();
  hCatPartCols.clear();
  for (ColumnGenerator gen : extraCols) {
    if (gen.getKeyType() == KeyType.STATIC_KEY) {
      hCatPartCols
        .add(new HCatFieldSchema(gen.getName(), gen.getHCatType(), ""));
      break;
    }
  }
  HCatSchema hCatPartSchema = new HCatSchema(hCatPartCols);
  return hCatPartSchema;
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
public String hCatRecordDump(List<HCatRecord> recs,
  HCatSchema schema) throws Exception {
  List<String> fields = schema.getFieldNames();
  int count = 0;
  StringBuilder sb = new StringBuilder(1024);
  for (HCatRecord rec : recs) {
    sb.append("HCat Record : " + ++count).append('\n');
    for (String field : fields) {
      sb.append('\t').append(field).append('=');
      sb.append(rec.get(field, schema)).append('\n');
      sb.append("\n\n");
    }
  }
  return sb.toString();
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
protected byte[] serializeStruct(HCatFieldSchema field, Map structData, int level) throws IOException {
    char separator = (char)separators[level];

    ByteArrayOutputStream baos = new ByteArrayOutputStream();

    HCatSchema structSchema = field.getStructSubSchema();

    boolean first = true;

    for(HCatFieldSchema structField: structSchema.getFields()) {
        if(!first) {
            baos.write(separator);
        }

        Object obj = structData.get(structField.getName());

        if(obj == null) {
            throw new IllegalArgumentException(
                    String.format("STRUCT types cannot have null members - field %s",
                            structField.getName()));
        }

        baos.write(serializeHiveType(structField, null, obj, level + 1));

        first = false;
    }

    return baos.toByteArray();
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
@Test
public void testSerializeMap() throws Exception {
    HCatTable table = mock(HCatTable.class);

    HCatSchema mapSchema = new HCatSchema(Lists.newArrayList(
            getSubFieldSchema("valueField", HCatFieldSchema.Type.BIGINT)));
    HCatFieldSchema fieldSchema = new HCatFieldSchema("testField",
            HCatFieldSchema.Type.MAP,
            HCatFieldSchema.Type.STRING,
            mapSchema, "");

    HiveSerializer serializer = new HiveSerializer(table);

    Map<String, Long> testMap = Maps.newLinkedHashMap();
    testMap.put("field1", 123L);
    testMap.put("field2", 456L);

    byte[] result = serializer.serializeMap(fieldSchema, testMap, 1);

    ByteArrayOutputStream expectedResult = new ByteArrayOutputStream();
    expectedResult.write(Bytes.toBytes("field1"));
    expectedResult.write('\003');
    expectedResult.write(Bytes.toBytes(123L));
    expectedResult.write('\002');
    expectedResult.write(Bytes.toBytes("field2"));
    expectedResult.write('\003');
    expectedResult.write(Bytes.toBytes(456L));

    byte[] expectedResultBytes = expectedResult.toByteArray();

    Assert.assertArrayEquals(expectedResultBytes, result);
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
public static void validateHCatTableAndTajoSchema(HCatSchema tblSchema) throws CatalogException {
  for (HCatFieldSchema hcatField : tblSchema.getFields()) {
    validateHCatFieldAndTajoSchema(hcatField);
  }
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
@Override
protected void setup(Context context)
  throws IOException, InterruptedException {
  Configuration conf = context.getConfiguration();
  String inputJobInfoStr = conf.get(HCatConstants.HCAT_KEY_JOB_INFO);
  jobInfo =
    (InputJobInfo) HCatUtil.deserialize(inputJobInfoStr);
  dataColsSchema = jobInfo.getTableInfo().getDataColumns();
  partitionSchema =
    jobInfo.getTableInfo().getPartitionColumns();
  StringBuilder storerInfoStr = new StringBuilder(1024);
  StorerInfo storerInfo = jobInfo.getTableInfo().getStorerInfo();
  storerInfoStr.append("HCatalog Storer Info : ")
    .append("\n\tHandler = ").append(storerInfo.getStorageHandlerClass())
    .append("\n\tInput format class = ").append(storerInfo.getIfClass())
    .append("\n\tOutput format class = ").append(storerInfo.getOfClass())
    .append("\n\tSerde class = ").append(storerInfo.getSerdeClass());
  Properties storerProperties = storerInfo.getProperties();
  if (!storerProperties.isEmpty()) {
    storerInfoStr.append("\nStorer properties ");
    for (Map.Entry<Object, Object> entry : storerProperties.entrySet()) {
      String key = (String) entry.getKey();
      Object val = entry.getValue();
      storerInfoStr.append("\n\t").append(key).append('=').append(val);
    }
  }
  storerInfoStr.append("\n");
  LOG.info(storerInfoStr);

  hCatFullTableSchema = new HCatSchema(dataColsSchema.getFields());
  for (HCatFieldSchema hfs : partitionSchema.getFields()) {
    hCatFullTableSchema.append(hfs);
  }
  fieldCount = hCatFullTableSchema.size();
  lobLoader = new LargeObjectLoader(conf,
    new Path(jobInfo.getTableInfo().getTableLocation()));
  bigDecimalFormatString = conf.getBoolean(
    ImportJobBase.PROPERTY_BIGDECIMAL_FORMAT,
    ImportJobBase.PROPERTY_BIGDECIMAL_FORMAT_DEFAULT);
  debugHCatImportMapper = conf.getBoolean(
    SqoopHCatUtilities.DEBUG_HCAT_IMPORT_MAPPER_PROP, false);
  IntWritable[] delimChars = DefaultStringifier.loadArray(conf,
      SqoopHCatUtilities.HIVE_DELIMITERS_TO_REPLACE_PROP, IntWritable.class);
  hiveDelimiters = new DelimiterSet(
    (char) delimChars[0].get(), (char) delimChars[1].get(),
    (char) delimChars[2].get(), (char) delimChars[3].get(),
    delimChars[4].get() == 1 ? true : false);
  hiveDelimsReplacement =
    conf.get(SqoopHCatUtilities.HIVE_DELIMITERS_REPLACEMENT_PROP);
  if (hiveDelimsReplacement == null) {
    hiveDelimsReplacement = "";
  }
  doHiveDelimsReplacement = Boolean.valueOf(conf.get(
    SqoopHCatUtilities.HIVE_DELIMITERS_REPLACEMENT_ENABLED_PROP));

  IntWritable[] fPos = DefaultStringifier.loadArray(conf,
      SqoopHCatUtilities.HCAT_FIELD_POSITIONS_PROP, IntWritable.class);
  hCatFieldPositions = new int[fPos.length];
  for (int i = 0; i < fPos.length; ++i) {
    hCatFieldPositions[i] = fPos[i].get();
  }

  LOG.debug("Hive delims replacement enabled : " + doHiveDelimsReplacement);
  LOG.debug("Hive Delimiters : " + hiveDelimiters.toString());
  LOG.debug("Hive delimiters replacement : " + hiveDelimsReplacement);
  staticPartitionKey =
    conf.get(SqoopHCatUtilities.HCAT_STATIC_PARTITION_KEY_PROP);
  LOG.debug("Static partition key used : " + staticPartitionKey);


}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
public HCatSchema getHCatOutputSchema() {
  return hCatOutputSchema;
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
public void setHCatOutputSchema(HCatSchema schema) {
  hCatOutputSchema = schema;
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
public HCatSchema getHCatPartitionSchema() {
  return hCatPartitionSchema;
}
 

import org.apache.hcatalog.data.schema.HCatSchema; //导入依赖的package包/类
public void setHCatPartitionSchema(HCatSchema schema) {
  hCatPartitionSchema = schema;
}