import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
private void flush() throws HCatException {
  if (hCatRecordsBatch.isEmpty()) {
    return;
  }
  try {
    slaveWriter.write(hCatRecordsBatch.iterator());
    masterWriter.commit(writerContext);
  } catch (HCatException e) {
    LOG.error("Exception in flush - write/commit data to Hive", e);
    //abort on exception
    masterWriter.abort(writerContext);
    throw e;
  } finally {
    hCatRecordsBatch.clear();
  }
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
/**
 * Specifies that the InputFormat returns Flink tuples instead of
 * {@link org.apache.hive.hcatalog.data.HCatRecord}.
 *
 * <p>Note: Flink tuples might only support a limited number of fields (depending on the API).
 *
 * @return This InputFormat.
 * @throws org.apache.hive.hcatalog.common.HCatException
 */
public HCatInputFormatBase<T> asFlinkTuples() throws HCatException {

	// build type information
	int numFields = outputSchema.getFields().size();
	if (numFields > this.getMaxFlinkTupleSize()) {
		throw new IllegalArgumentException("Only up to " + this.getMaxFlinkTupleSize() +
				" fields can be returned as Flink tuples.");
	}

	TypeInformation[] fieldTypes = new TypeInformation[numFields];
	fieldNames = new String[numFields];
	for (String fieldName : outputSchema.getFieldNames()) {
		HCatFieldSchema field = outputSchema.get(fieldName);

		int fieldPos = outputSchema.getPosition(fieldName);
		TypeInformation fieldType = getFieldType(field);

		fieldTypes[fieldPos] = fieldType;
		fieldNames[fieldPos] = fieldName;

	}
	this.resultType = new TupleTypeInfo(fieldTypes);

	return this;
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
/**
 * Specifies that the InputFormat returns Flink tuples instead of
 * {@link org.apache.hive.hcatalog.data.HCatRecord}.
 *
 * Note: Flink tuples might only support a limited number of fields (depending on the API).
 *
 * @return This InputFormat.
 * @throws org.apache.hive.hcatalog.common.HCatException
 */
public HCatInputFormatBase<T> asFlinkTuples() throws HCatException {

	// build type information
	int numFields = outputSchema.getFields().size();
	if(numFields > this.getMaxFlinkTupleSize()) {
		throw new IllegalArgumentException("Only up to "+this.getMaxFlinkTupleSize()+
				" fields can be returned as Flink tuples.");
	}

	TypeInformation[] fieldTypes = new TypeInformation[numFields];
	fieldNames = new String[numFields];
	for (String fieldName : outputSchema.getFieldNames()) {
		HCatFieldSchema field = outputSchema.get(fieldName);

		int fieldPos = outputSchema.getPosition(fieldName);
		TypeInformation fieldType = getFieldType(field);

		fieldTypes[fieldPos] = fieldType;
		fieldNames[fieldPos] = fieldName;

	}
	this.resultType = new TupleTypeInfo(fieldTypes);

	return this;
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
private void insert(Map<String, String> partitionSpec, Iterable<HCatRecord> rows) {
  WriteEntity entity = new WriteEntity.Builder()
      .withDatabase(databaseName)
      .withTable(tableName)
      .withPartition(partitionSpec)
      .build();

  try {
    HCatWriter master = DataTransferFactory.getHCatWriter(entity, config);
    WriterContext context = master.prepareWrite();
    HCatWriter writer = DataTransferFactory.getHCatWriter(context);
    writer.write(rows.iterator());
    master.commit(context);
  } catch (HCatException e) {
    throw new RuntimeException("An error occurred while inserting data to " + databaseName + "." + tableName, e);
  }
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
private ReaderContext getReaderContext(long desiredSplitCount) throws HCatException {
  ReadEntity entity =
      new ReadEntity.Builder()
          .withDatabase(spec.getDatabase())
          .withTable(spec.getTable())
          .withFilter(spec.getFilter())
          .build();
  // pass the 'desired' split count as an hint to the API
  Map<String, String> configProps = new HashMap<>(spec.getConfigProperties());
  configProps.put(
      HCatConstants.HCAT_DESIRED_PARTITION_NUM_SPLITS, String.valueOf(desiredSplitCount));
  return DataTransferFactory.getHCatReader(entity, configProps).prepareRead();
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
@Override
public boolean start() throws HCatException {
  HCatReader reader =
      DataTransferFactory.getHCatReader(source.spec.getContext(), source.spec.getSplitId());
  hcatIterator = reader.read();
  return advance();
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
@Setup
public void initiateWrite() throws HCatException {
  WriteEntity entity =
      new WriteEntity.Builder()
          .withDatabase(spec.getDatabase())
          .withTable(spec.getTable())
          .withPartition(spec.getPartition())
          .build();
  masterWriter = DataTransferFactory.getHCatWriter(entity, spec.getConfigProperties());
  writerContext = masterWriter.prepareWrite();
  slaveWriter = DataTransferFactory.getHCatWriter(writerContext);
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
@ProcessElement
public void processElement(ProcessContext ctx) throws HCatException {
  hCatRecordsBatch.add(ctx.element());
  if (hCatRecordsBatch.size() >= spec.getBatchSize()) {
    flush();
  }
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
private static void validateHcatFieldFollowsPigRules(HCatFieldSchema hcatField)
    throws PigException {
  try {
    Type hType = hcatField.getType();
    switch (hType) {
    case BOOLEAN:
      if (!pigHasBooleanSupport) {
        throw new PigException("Incompatible type found in HCat table schema: "
            + hcatField, PigHCatUtil.PIG_EXCEPTION_CODE);
      }
      break;
    case ARRAY:
      validateHCatSchemaFollowsPigRules(hcatField.getArrayElementSchema());
      break;
    case STRUCT:
      validateHCatSchemaFollowsPigRules(hcatField.getStructSubSchema());
      break;
    case MAP:
      // key is only string
      if (hcatField.getMapKeyType() != Type.STRING) {
        LOG.info("Converting non-String key of map " + hcatField.getName() + " from "
          + hcatField.getMapKeyType() + " to String.");
      }
      validateHCatSchemaFollowsPigRules(hcatField.getMapValueSchema());
      break;
    }
  } catch (HCatException e) {
    throw new PigException("Incompatible type found in hcat table schema: " + hcatField,
        PigHCatUtil.PIG_EXCEPTION_CODE, e);
  }
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
private Object get(String name) {
  checkColumn(name);
  try {
    return row.get(name, schema);
  } catch (HCatException e) {
    throw new RuntimeException("Error getting value for " + name, e);
  }
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
private static HCatFieldSchema column(String name) {
  try {
    return new HCatFieldSchema(name, STRING, null);
  } catch (HCatException e) {
    throw new RuntimeException(e);
  }
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
@FinishBundle
public void finishBundle() throws HCatException {
  flush();
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
/** Returns a ReaderContext instance for the passed datastore config params. */
static ReaderContext getReaderContext(Map<String, String> config) throws HCatException {
  return DataTransferFactory.getHCatReader(READ_ENTITY, config).prepareRead();
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
/** Returns a WriterContext instance for the passed datastore config params. */
private static WriterContext getWriterContext(Map<String, String> config) throws HCatException {
  return DataTransferFactory.getHCatWriter(WRITE_ENTITY, config).prepareWrite();
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
/** Writes records to the table using the passed WriterContext. */
private static void writeRecords(WriterContext context) throws HCatException {
  DataTransferFactory.getHCatWriter(context).write(getHCatRecords(TEST_RECORDS_COUNT).iterator());
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
private static Iterator<HCatRecord> loadHCatRecordItr(ReaderContext readCntxt, int dataSplit) throws HCatException {
    HCatReader currentHCatReader = DataTransferFactory.getHCatReader(readCntxt, dataSplit);

    return currentHCatReader.read();
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
private static Iterator<HCatRecord> loadHCatRecordItr(ReaderContext readCntxt, int dataSplit) throws HCatException {
    HCatReader currentHCatReader = DataTransferFactory.getHCatReader(readCntxt, dataSplit);
    return currentHCatReader.read();
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
private HCatSchema buildTestSchema() throws HCatException {
    HCatSchema hCatSchema = HCatSchemaUtils.getListSchemaBuilder().build();
    String[] fields = new String[] {
            "general_transaction_id",
            "program_year",
            "payment_publication_date",
            "submitting_applicable_manufacturer_or_applicable_gpo_name",
            "covered_recipient_type",
            "teaching_hospital_id",
            "teaching_hospital_name",
            "physician_profile_id",
            "physician_first_name",
            "physician_middle_name",
            "physician_last_name",
            "physician_name_suffix",
            "recipient_primary_business_street_address_line1",
            "recipient_primary_business_street_address_line2",
            "recipient_city",
            "recipient_state",
            "recipient_zip_code",
            "recipient_country",
            "recipient_province",
            "recipient_postal_code",
            "physician_primary_type",
            "physician_specialty",
            "physician_license_state_code1",
            "physician_license_state_code2",
            "physician_license_state_code3",
            "physician_license_state_code4",
            "physician_license_state_code5",
            "product_indicator",
            "name_of_associated_covered_drug_or_biological1",
            "name_of_associated_covered_drug_or_biological2",
            "name_of_associated_covered_drug_or_biological3",
            "name_of_associated_covered_drug_or_biological4",
            "name_of_associated_covered_drug_or_biological5",
            "ndc_of_associated_covered_drug_or_biological1",
            "ndc_of_associated_covered_drug_or_biological2",
            "ndc_of_associated_covered_drug_or_biological3",
            "ndc_of_associated_covered_drug_or_biological4",
            "ndc_of_associated_covered_drug_or_biological5",
            "name_of_associated_covered_device_or_medical_supply1",
            "name_of_associated_covered_device_or_medical_supply2",
            "name_of_associated_covered_device_or_medical_supply3",
            "name_of_associated_covered_device_or_medical_supply4",
            "name_of_associated_covered_device_or_medical_supply5",
            "applicable_manufacturer_or_applicable_gpo_making_payment_name",
            "applicable_manufacturer_or_applicable_gpo_making_payment_id",
            "applicable_manufacturer_or_applicable_gpo_making_payment_state",
            "applicable_manufacturer_or_applicable_gpo_making_payment_country",
            "dispute_status_for_publication",
            "total_amount_of_payment_usdollars",
            "date_of_payment",
            "number_of_payments_included_in_total_amount",
            "form_of_payment_or_transfer_of_value",
            "nature_of_payment_or_transfer_of_value",
            "city_of_travel",
            "state_of_travel",
            "country_of_travel",
            "physician_ownership_indicator",
            "third_party_payment_recipient_indicator",
            "name_of_third_party_entity_receiving_payment_or_transfer_of_value",
            "charity_indicator",
            "third_party_equals_covered_recipient_indicator",
            "contextual_information",
            "delay_in_publication_of_general_payment_indicator" };
    for (String field : fields) {
        hCatSchema.append(new HCatFieldSchema(field, TypeInfoFactory.stringTypeInfo, ""));
    }
    return hCatSchema;
}
 

import org.apache.hive.hcatalog.common.HCatException; //导入依赖的package包/类
protected abstract T buildFlinkTuple(T t, HCatRecord record) throws HCatException;