Colocation

功能简介

同分布（Colocation）功能是将存在关联关系的数据或可能要进行关联操作的数据存储在相同的存储节点上。HDFS文件同分布的特性，将那些需进行关联操作的文件存放在相同数据节点上，在进行关联操作计算时避免了到别的数据节点上获取数据，大大降低网络带宽的占用。

在使用Colocation功能之前，建议用户对Colocation的内部机制有一定了解，包括：

• Colocation分配节点原理

• 扩容与Colocation分配

• Colocation与数据节点容量

• Colocation分配节点原理 Colocation为locator分配数据节点的时候，locator的分配算法会根据已分配的情况，进行均衡的分配数据节点。

  说明： locator分配算法的原理是，查询目前存在的所有locators，读取所有locators所分配的数据节点，并记录其使用次数。根据使用次数，对数据节点进行排序，使用次数少的排在前面，优先选择排在前面的节点。每次选择一个节点后，计数加1，并重新排序，选择后续的节点。

• 扩容与Colocation分配 集群扩容之后，为了平衡地使用所有的数据节点，使新的数据节点的分配频率与旧的数据节点趋于一致，有如下两种策略可以选择，如表1所示。

表1 分配策略

说明：

• 一般的，建议用户在进行集群扩容之后采用策略一来重新分配locators，可以避免数据偏重使用新的数据节点。

• Colocation与数据节点容量 由于使用Colocation进行存储数据的时候，会固定存储在指定的locator所对应的数据节点上面，所以如果不对locator进行规划，会造成数据节点容量不均衡。下面总结了保证数据节点容量均衡的两个主要的使用原则，如表2所示。

表2 使用原则

HDFS的二次开发过程中，可以获取DFSColocationAdmin和DFSColocationClient实例，进行从location创建group、删除group、写文件和删除文件的操作。

说明：

• 使用Colocation功能，用户指定了DataNode，会造成某些节点上数据量很大。数据倾斜严重，导致HDFS写任务失败。
• 由于数据倾斜，导致MapReduce只会在某几个节点访问，造成这些节点上负载很大，而其他节点闲置。
• 针对单个应用程序任务，只能使用一次DFSColocationAdmin和DFSColocationClient实例。如果每次对文件系统操作都获取此实例，会创建过多HDFS链接，消耗HDFS资源。
• 如果需要对colocation上传的文件做balance操作，为避免colocation失效，可以通过FusionInsight Manager界面中的oi.dfs.colocation.file.pattern参数进行设置，设置该参数值为对应数据文件块的路径，多个路径之间以逗号分开。例如/test1，/test2。

代码样例

完整样例代码可参考com.huawei.bigdata.hdfs.examples.ColocationExample。

说明：

在运行Colocation工程时，需要将HDFS用户绑定supergroup用户组。

1、初始化

使用Colocation前需要进行kerberos安全认证。

private static void init() throws IOException {
    LoginUtil.login(PRNCIPAL_NAME, PATH_TO_KEYTAB, PATH_TO_KRB5_CONF, conf);
  }

2、获取实例

样例：Colocation的操作使用DFSColocationAdmin和DFSColocationClient实例，在进行创建group等操作前需获取实例。

public static void main(String[] args) throws IOException {
    init();
    dfsAdmin = new DFSColocationAdmin(conf);
    dfs = new DFSColocationClient();
    dfs.initialize(URI.create("hdfs://hacluster"), conf);
    createGroup();
    put();
    delete();
    deleteGroup();
    dfs.close();
    dfsAdmin.close();
  }

3、创建group

样例：创建一个gid01组，组中包含3个locator。

  private static void createGroup() throws IOException {
    dfsAdmin.createColocationGroup(COLOCATION_GROUP_GROUP01,
        Arrays.asList(new String[] { "lid01", "lid02", "lid03" }));
  }

4、写文件，写文件前必须创建对应的group

样例：写入testfile.txt文件。

 private static void put() throws IOException {
    FSDataOutputStream out = dfs.create(new Path("/testfile.txt"), true,
        COLOCATION_GROUP_GROUP01, "lid01");
    // 待写入到HDFS的数据.
    byte[] readBuf = "Hello World".getBytes("UTF-8");
    out.write(readBuf, 0, readBuf.length);
    out.close();
  }

5、删除文件

样例：删除testfile.txt文件。

 public static void delete() throws IOException {
    dfs.delete(new Path("/testfile.txt"));
  }

6、删除group

样例：删除gid01。

 private static void deleteGroup() throws IOException {
    dfsAdmin.deleteColocationGroup(COLOCATION_GROUP_GROUP01);
  }