@VisibleForTesting
  static void writeDescriptor(
      FileSystem fs, Path metadataLocation, String name,
      DatasetDescriptor descriptor) {

    checkExists(fs, metadataLocation);

    // write the schema to the previous file location so
    // it can be read by earlier versions of Kite
    FSDataOutputStream outputStream = null;
    final Path schemaPath = new Path(metadataLocation, SCHEMA_FILE_NAME);
    boolean threw = true;
    try {
      outputStream = fs.create(schemaPath, true /* overwrite */ );
      outputStream.write(descriptor.getSchema().toString(true)
          .getBytes(Charsets.UTF_8));
      outputStream.flush();
      threw = false;
    } catch (IOException e) {
      throw new DatasetIOException(
          "Unable to save schema file: " + schemaPath +
          " for dataset: " + name, e);
    } finally {
      try {
        Closeables.close(outputStream, threw);
      } catch (IOException e) {
        throw new DatasetIOException("Cannot close", e);
      }
    }

    // use the SchemaManager for schema operations moving forward
    SchemaManager manager = SchemaManager.create(fs.getConf(),
        new Path(metadataLocation, SCHEMA_DIRECTORY_NAME));

    manager.writeSchema(descriptor.getSchema());

    Properties properties = new Properties();
    properties.setProperty(VERSION_FIELD_NAME, METADATA_VERSION);
    properties.setProperty(FORMAT_FIELD_NAME, descriptor.getFormat().getName());
    properties.setProperty(COMPRESSION_TYPE_FIELD_NAME, descriptor.getCompressionType().getName());

    final URI dataLocation = descriptor.getLocation();
    if (dataLocation != null) {
      properties.setProperty(LOCATION_FIELD_NAME, dataLocation.toString());
    }

    if (descriptor.isPartitioned()) {
      properties.setProperty(PARTITION_EXPRESSION_FIELD_NAME,
          Accessor.getDefault().toExpression(descriptor.getPartitionStrategy()));
    }

    // copy custom properties to the table
    for (String property : descriptor.listProperties()) {
      // no need to check the reserved list, those are not set on descriptors
      properties.setProperty(property, descriptor.getProperty(property));
    }

    final Path descriptorPath = new Path(metadataLocation, DESCRIPTOR_FILE_NAME);
    threw = true;
    try {
      outputStream = fs.create(descriptorPath, true /* overwrite */ );
      properties.store(outputStream, "Dataset descriptor for " + name);
      outputStream.flush();
      threw = false;
    } catch (IOException e) {
      throw new DatasetIOException(
          "Unable to save descriptor file: " + descriptorPath +
          " for dataset: " + name, e);
    } finally {
      try {
        Closeables.close(outputStream, threw);
      } catch (IOException e) {
        throw new DatasetIOException("Cannot close", e);
      }
    }
  }
@ VisibleForTesting static void writeDescriptor ( FileSystem fs , Path metadataLocation , String name , DatasetDescriptor descriptor ) { checkExists ( fs , metadataLocation ) ; FSDataOutputStream outputStream = null ; final Path schemaPath = new Path ( metadataLocation , SCHEMA_FILE_NAME ) ; boolean threw = true ; try { outputStream = fs . create ( schemaPath , true ) ; outputStream . write ( descriptor . getSchema ( ) . toString ( true ) . getBytes ( Charsets . UTF_8 ) ) ; outputStream . flush ( ) ; threw = false ; } catch ( IOException e ) { throw new DatasetIOException ( "Unable to save schema file: " + schemaPath + " for dataset: " + name , e ) ; } finally { try { Closeables . close ( outputStream , threw ) ; } catch ( IOException e ) { throw new DatasetIOException ( "Cannot close" , e ) ; } } SchemaManager manager = SchemaManager . create ( fs . getConf ( ) , new Path ( metadataLocation , SCHEMA_DIRECTORY_NAME ) ) ; manager . writeSchema ( descriptor . getSchema ( ) ) ; Properties properties = new Properties ( ) ; properties . setProperty ( VERSION_FIELD_NAME , METADATA_VERSION ) ; properties . setProperty ( FORMAT_FIELD_NAME , descriptor . getFormat ( ) . getName ( ) ) ; properties . setProperty ( COMPRESSION_TYPE_FIELD_NAME , descriptor . getCompressionType ( ) . getName ( ) ) ; final URI dataLocation = descriptor . getLocation ( ) ; if ( dataLocation != null ) { properties . setProperty ( LOCATION_FIELD_NAME , dataLocation . toString ( ) ) ; } if ( descriptor . isPartitioned ( ) ) { properties . setProperty ( PARTITION_EXPRESSION_FIELD_NAME , Accessor . getDefault ( ) . toExpression ( descriptor . getPartitionStrategy ( ) ) ) ; } for ( String property : descriptor . listProperties ( ) ) { properties . setProperty ( property , descriptor . getProperty ( property ) ) ; } final Path descriptorPath = new Path ( metadataLocation , DESCRIPTOR_FILE_NAME ) ; threw = true ; try { outputStream = fs . create ( descriptorPath , true ) ; properties . store ( outputStream , "Dataset descriptor for " + name ) ; outputStream . flush ( ) ; threw = false ; } catch ( IOException e ) { throw new DatasetIOException ( "Unable to save descriptor file: " + descriptorPath + " for dataset: " + name , e ) ; } finally { try { Closeables . close ( outputStream , threw ) ; } catch ( IOException e ) { throw new DatasetIOException ( "Cannot close" , e ) ; } } }
Writes the contents of a {@code Descriptor} to files.

@param fs                The {@link FileSystem} where data will be stored
@param metadataLocation  The directory {@link Path} where metadata files
will be located
@param name              The {@link Dataset} name
@param descriptor        The {@code Descriptor} contents to write

@throws org.kitesdk.data.DatasetIOException
If the {@code metadataLocation} does not exist or
if any IOExceptions need to be propagated.