With the release of version 3.0.0, Hadoop introduced the use of Erasure Coding (EC) in HDFS. By default HDFS achieves durability via block replication. Usually the replication count is 3, resulting in a storage overhead of 200%. Hadoop 3 introduced EC as a better way to achieve durability. EC behaves much like RAID 5 or 6…for k blocks of data, m blocks of parity data are generated, from which the original data can be recovered in the event of disk or node failures (erasures, in EC parlance). A typical EC scheme is Reed-Solomon 6-3, where 6 data blocks produce 3 parity blocks, an overhead of only 50%. In addition to doubling the available disk space, RS-6-3 is also more fault tolerant…a loss of 3 data blocks can be tolerated, whereas triple replication can only sustain a loss of two.

To use EC with Accumulo, it is highly recommended that you first rebuild Hadoop with support for Intel’s ISA-L library. Instructions for doing this can be found here

Important Warning

As noted here, the current EC implementation does not support hflush() and hsync(). These functions are no-ops, which means that EC coded files are not guaranteed to be written to disk after a sync or flush. For this reason, EC should never be used for the Accumulo write-ahead logs. Data loss may, and most likely will, occur. It is also recommended that tables in the accumulo namespace (root and metadata for example) continue to use replication.

EC and Threads

Due to the striped nature of an EC encoded file, an EC enabled HDFS client is threaded. This becomes an issue when an Accumulo client or service is configured to use multiple threads to read or write to HDFS, and becomes especially problematic when doing bulk imports. By default, Accumulo will use eight times the number of cores on the client machine to scan the files to be imported and map them to tablet files. Each thread created to scan the input files will create on the order of k threads to perform parallel I/O. RS-10-4 on a 16 core machine, for instance, will spawn over a thousand threads to perform this operation. If sufficient memory is not available, this operation will fail without providing a meaningful error message to the user. This particular problem can be ameliorated by setting the bulk.threads client property to 1C (i.e. one thread per core), down from the default of 8C. Similar care should be taken when setting other thread limits.

HDFS ec Command

Encoding policy in HDFS is set at the directory level, with children inheriting policies from their parents if not explicitly set. The encoding policy for a directory can be manipulated via the hdfs ec command, documented here.

The first step is to determine which policies are configured for your HDFS instance. This is done via the -listPolicies command. The following listing shows that there are 5 configured policies, of which only 3 (RS-10-4-1024k, RS-6-3-1024k, and RS-6-3-64k) are enabled for use.

  1. $ hdfs ec -listPolicies
  2. Erasure Coding Policies:
  3. ErasureCodingPolicy=[Name=RS-10-4-1024k, Schema=[ECSchema=[Codec=rs, numDataUnits=10, numParityUnits=4]], CellSize=1048576, Id=5], State=ENABLED
  4. ErasureCodingPolicy=[Name=RS-6-3-1024k, Schema=[ECSchema=[Codec=rs, numDataUnits=6, numParityUnits=3]], CellSize=1048576, Id=1], State=ENABLED
  5. ErasureCodingPolicy=[Name=RS-6-3-64k, Schema=[ECSchema=[Codec=rs, numDataUnits=6, numParityUnits=3, options=]], CellSize=65536, Id=65], State=ENABLED
  6. ErasureCodingPolicy=[Name=RS-LEGACY-6-3-1024k, Schema=[ECSchema=[Codec=rs-legacy, numDataUnits=6, numParityUnits=3]], CellSize=1048576, Id=3], State=DISABLED
  7. ErasureCodingPolicy=[Name=XOR-2-1-1024k, Schema=[ECSchema=[Codec=xor, numDataUnits=2, numParityUnits=1]], CellSize=1048576, Id=4], State=DISABLED

To set the encoding policy for a directory, use the -setPolicy command.

  1. $ hadoop fs -mkdir foo
  2. $ hdfs ec -setPolicy -policy RS-6-3-64k -path foo
  3. Set RS-6-3-64k erasure coding policy on foo

To get the encoding policy for a directory, use the -getPolicy command.

  1. $ hdfs ec -getPolicy -path foo
  2. RS-6-3-64k

New directories created under foo will inherit the EC policy.

  1. $ hadoop fs -mkdir foo/bar
  2. $ hdfs ec -getPolicy -path foo/bar
  3. RS-6-3-64k

And changing the policy for a parent will also change its children. The -setPolicy command here issues a warning that existing files will not be converted. To switch the policy for an existing file, you must create a new file (through a copy, for instance). For Accumulo, if you change the encoding policy for a table’s directories, you would then have to perform a major compaction on the table to convert the table’s RFiles to the desired encoding.

  1. $ hdfs ec -setPolicy -policy RS-6-3-1024k -path foo
  2. Set RS-6-3-1024k erasure coding policy on foo
  3. Warning: setting erasure coding policy on a non-empty directory will not automatically convert existing files to RS-6-3-1024k erasure coding policy
  4. $ hdfs ec -getPolicy -path foo
  5. RS-6-3-1024k
  6. $ hdfs ec -getPolicy -path foo/bar
  7. RS-6-3-1024k

Configuring EC for a New Instance

If you wish to create a new instance with a single encoding policy for all tables, you simply need to change the encoding policy on the tables directory after running accumulo init (see Quick Start guide). To keep the tables in the accumulo namespace using replication, you would then need to manually change them back to using replication. Assuming Accumulo is configured to use /accumulo as its root, you would do the following:

  1. $ hdfs ec -setPolicy -policy RS-6-3-64k -path /accumulo/tables
  2. Set RS-6-3-64k erasure coding policy on /accumulo/tables
  3. $ hdfs ec -setPolicy -replicate -path /accumulo/tables/\!0
  4. Set replication erasure coding policy on /accumulo/tables/!0
  5. $ hdfs ec -setPolicy -replicate -path /accumulo/tables/+r
  6. Set replication erasure coding policy on /accumulo/tables/+r
  7. $ hdfs ec -setPolicy -replicate -path /accumulo/tables/+rep
  8. Set replication erasure coding policy on /accumulo/tables/+rep

Check that the policies are set correctly:

  1. $ hdfs ec -getPolicy -path /accumulo/tables
  2. RS-6-3-64k
  3. $ hdfs ec -getPolicy -path /accumulo/tables/\!0
  4. The erasure coding policy of /accumulo/tables/!0 is unspecified

Any directories subsequently created under /accumulo/tables will be erasure coded.

Configuring EC for an Existing Instance

For an existing installation, the instructions are the same, but with the caveat that changing the encoding policy for an existing directory will not change the files within the directory. Converting existing tables to EC requires a major compaction to complete the process. For instance, to convert test.table1 to RS-6-3-64k, you would first find the table ID via the accumulo shell, use hdfs ec to change the encoding for the directory /accumulo/tables/<tableID>, and then compact the table.

  1. $ accumulo shell
  2. user@instance> tables -l
  3. accumulo.metadata    =>        !0
  4. accumulo.replication =>      +rep
  5. accumulo.root        =>        +r
  6. test.table1          =>         3
  7. test.table2          =>         4
  8. test.table3          =>         5
  9. trace                =>         1
  10. user@instance> quit
  11. $ hdfs ec -setPolicy -policy RS-6-3-64k -path /accumulo/tables/3
  12. Set RS-6-3-64k erasure coding policy on /accumulo/tables/3
  13. $ accumulo shell
  14. user@instance> compact -t test.table1

Defining Custom EC Policies

Hadoop by default will enable only a single EC policy, which is determined by the value of the dfs.namenode.ec.system.default.policy configuration setting. To enable an existing policy, use the hdfs ec -enablePolicy command. To define custom policies, you must first edit the user_ec_policies.xml file found in the Hadoop configuration directory, and then run the hdfs ec -addPolicies command. For example, to add RS-6-3-64k as a policy, you first edit user_ec_policies.xml and add the following:

  1. <configuration>
  2. <layoutversion>1</layoutversion>
  3. <schemas>
  4.   <!-- schema id is only used to reference internally in this document -->
  5.   <schema id="RSk6m3">
  6.     <codec>rs</codec>
  7.     <k>6</k>
  8.     <m>3</m>
  9.     <options> </options>
  10.   </schema>
  11. </schemas>
  12. <policies>
  13.   <policy>
  14.     <schema>RSk6m3</schema>
  15.     <cellsize>65536</cellsize>
  16.   </policy>
  17. </policies>
  18. </configuration>

Here the schema “RSk6m3” defines a Reed-Solomon encoding with k\=6 data blocks and m\=3 parity blocks. This schema is then used to define a policy that uses RS-6-3 encoding with a stripe size of 64k. To add this policy:

  1. $ hdfs ec -addPolicies -policyFile /hadoop/etc/hadoop/user_ec_policies.xml
  2. 2019-11-19 15:35:23,703 INFO util.ECPolicyLoader: Loading EC policy file /hadoop/etc/hadoop/user_ec_policies.xml
  3. Add ErasureCodingPolicy RS-6-3-64k succeed.

To enable the policy:

  1. $ hdfs ec -enablePolicy -policy RS-6-3-64k
  2. Erasure coding policy RS-6-3-64k is enabled