sqoop-union-avro.scala

/**
 * Generate an avro schema for a database.
 *
 * Uses Sqoop's AvroGenerator but switchs the unions to be [null, type]
 * and sets default to null
 *
 * @param db_type Type of database
 * @param options The SqoopOptions the AvroGenerator will use.
 * @return A avro schema representing the sqoop table
 * */
def get_avro_schema(db_type : String, options: SqoopOptions) : Schema = {
  val table_name = options.getTableName
  val conn = db_type match {
    case "mysql" =>
      if (options.isDirect) new MySQLManager(options) else new DirectMySQLManager(options)
    case "netezza" =>
      if (options.isDirect) new NetezzaManager(options) else new DirectNetezzaManager(options)
  }
  val avro = new AvroSchemaGenerator(options, conn, options.getTableName)
  log.info("Connecting to %s to generate Avro schema for ".format(options.getConnectString, table_name))
  val schema = avro.generate()
 
  //now add default value (set to null for sqoop)
  val fields = schema.getFields
  var new_fields = new ListBuffer[Field]()
 
  fields.foreach(field => {
    //switch union to be [null, type]. What do we set as the default otherwise, this way can set null
    val new_types : util.List[Schema] = List(Schema.create(Schema.Type.NULL), Schema.create(get_non_null(field.schema()).getType))
    val new_field : Field = new Field(field.name(), Schema.createUnion(new_types), null, NullNode.getInstance())
    new_fields += new_field
  })
 
  //recreate sqoop attributes
  val doc : String = "Sqoop import of " + table_name
  val new_schema : Schema = Schema.createRecord(table_name, doc, null, false)
  new_schema.setFields(new_fields)
  new_schema
}