hadoop实现原理(一)配置信息处理

Hadoop作为一个复杂的软件系统，使用一个配置模块提高其适应性或扩展性，作为其扩展、定制的手段和方式。

为什么不直接使用java.util.Properties类？

Properties类继承自Hashtable，它并不能支持INI文件的“节”，对配置项进行分类。Properties类提供了load()方法加载，该方法从输入流中读入key-value对，而store()方法则将Properties表中的属性列表写入输出流。还提供了loadFromXML()和storeToXML()方法。相应get和set方法：

public String getProperty(String key)

public String getProperty(String key, String defaultValue)

public synchronized Object setProperty(String key, String value)

由于java.util.Properties提供的能力有限，Java社区出现了大量的配置信息读写方案，其中比较著名的是Apache Jakarta Commons工具集中提供的Commons Configuration。

Hadoop没有使用Properties类，也没有使用Commons Configuration来管理配置文件，而使用了一套独有的配置文件管理系统，并提供自己的API，即使用org.apache.hadoop.conf.Configuration。

Hadoop配置文件的格式

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<configuration>
<property>
<name>hadoop.tmp.dir</name>
<value>/tmp/hadoop-${user.name}</value>
</property>
<property>
<name>fs.default.name</name>
<value>hdfs://localhost:54310</value>
</property>
<property>
<name>mapred.job.tracker</name>
<value>hdfs://localhost:54311</value>
</property>
<property>
<name>dfs.replication</name>
<value>8</value>
</property>
<property>
<name>mapred.child.java.opts</name>
<value>-Xmx512m</value>
</property>
</configuration>

Hadoop配置文件的根元素是configuration，一般只包含子元素property。Hadoop可以合并资源将多个配置文件合并，产生一个配置。如下：

Configurationconf = new Configuration();

conf.addResource(“core-default.xml”);

conf.addResource(“core-site.xml”)

如果这两个资源都包含了相同的配置项，而且前一个资源的配置项没有标记为final，那么，后面的配置将覆盖前面的配置。如果第一个资源中某配置项标记了final,那么，在加载第二个资源的时候，会有警告提示。

Hadoop配置系统还有一个重要的功能，就是属性扩展。如配置项项dfs.name.dir的值是Hadoop.tmp.dir/dfs/name，其中，{Hadoop.tmp.dir}会使用Configuration中的相应属性值进行扩展。

Configuration的成员变量



使用Configuration类的一般过程是：构造Configuration对象，并通过类的addResource()方法添加需要加载的资源：然后就可以用get*方法和set*方法访问和设置配置项。

quitemode，用来设置加载配置的模式，quite=true则不输出日志信息，方便调试。

Resources保存了所有通过addResource()方法添加Configuration对象的资源。Configuration.addResource()有如下4种形式：

public void addResource(InputStream in)
public void addResource(Path file)
public void addResource(String name)
public void addResource(URL url)

在HDFS的DataNode中，加载两个默认资源:

// 代码来自org.apache.hadoop.hdfs.server.datanode.DataNode

static{

Configuration.addDefaultResource(“hdfs-default.xml”);

Configuration.addDefaultResource(“hdfs-site.xml”)

}

properties和overlay成员变量都是前面介绍过的java.util.Properties，Hadoop配置文件解析后的key-value对都存在properties中，变量finalParameters的类型是Set，用来保存所有在配置文件中已经声明为fianl的键，变量overlay用于记录通过set()方式改变的配置项，也就是应用设置的。

资源加载

资源通过对象的addResource()方法或类的静态addDefaultResource()方法(设置了loadDefault标志)添加到Configuration对象中，添加的资源并不会立即被加载，只是通过reloadConfiguration()方法清空properties和finalParameters。相关代码如下:

public void addResource(String name){
addResourceObject(name);
}
private synchronized void addResourceObject(Object resource) {
Resources.add(resource);
reloadConfiguration();
}
public synchronized void reloadConfiguration(){
Properties = null;
finalParameters.clear();
}

静态方法addDefaultResource()也能清空Configuration对象中的数据，这是通过类的静态成员REGISTRY作为媒介进行的。

public static synchronized void addDefaultResource(String name) {
if(!defaultResources.contains(name)) {
defaultResources.add(name);
for(Configuration conf : REGISTRY.keySet()) {
if(conf.loadDefaults) {
conf.reloadConfiguration();
}
}
}
}

成员变量properties中的数据，直到需要的时候才会加载进来。在getProps()方法中，如果发现properties为空，将触发loadResources()方法加载配置资源。延迟加载的设计模式，当真需要配置数据的时候，才开始解析配置文件。

private synchronized Properties getProps() {
if (properties == null) {
properties = new Properties();
loadResources(properties, resources, quietmode);
if (overlay!= null) {
properties.putAll(overlay);
if (storeResource) {
for (Map.Entry<Object,Object> item: overlay.entrySet()) {
updatingResource.put((String) item.getKey(), "Unknown");
}
}
}
}
return properties;
}

由于Hadoop的配置文件都是很小的文件，因此Configuration使用DOM处理XML。

private void loadResources(Properties properties,
ArrayList resources,
boolean quiet) {
if(loadDefaults) {
for (String resource : defaultResources) {
loadResource(properties, resource, quiet);
}

//support the hadoop-site.xml as a deprecated case
if(getResource("hadoop-site.xml")!=null) {
loadResource(properties, "hadoop-site.xml", quiet);
}
}

for (Object resource : resources) {
loadResource(properties, resource, quiet);
}
}
private void loadResource(Properties properties, Object name, boolean quiet) {
try {
DocumentBuilderFactory docBuilderFactory
= DocumentBuilderFactory.newInstance();
//ignore all comments inside the xml file
docBuilderFactory.setIgnoringComments(true);

//allow includes in the xml file
docBuilderFactory.setNamespaceAware(true);
try {
docBuilderFactory.setXIncludeAware(true);
} catch (UnsupportedOperationException e) {
LOG.error("Failed to set setXIncludeAware(true) for parser "
+ docBuilderFactory
+ ":" + e,
e);
}
DocumentBuilder builder = docBuilderFactory.newDocumentBuilder();
Document doc = null;
Element root = null;

if (name instanceof URL) {                  // an URL resource
URL url = (URL)name;
if (url != null) {
if (!quiet) {
LOG.info("parsing " + url);
}
doc = builder.parse(url.toString());
}
} else if (name instanceof String) {        // a CLASSPATH resource
URL url = getResource((String)name);
if (url != null) {
if (!quiet) {
LOG.info("parsing " + url);
}
doc = builder.parse(url.toString());
}
} else if (name instanceof Path) {          // a file resource
// Can't use FileSystem API or we get an infinite loop
// since FileSystem uses Configuration API.  Use java.io.File instead.
File file = new File(((Path)name).toUri().getPath())
.getAbsoluteFile();
if (file.exists()) {
if (!quiet) {
LOG.info("parsing " + file);
}
InputStream in = new BufferedInputStream(new FileInputStream(file));
try {
doc = builder.parse(in);
} finally {
in.close();
}
}
} else if (name instanceof InputStream) {
try {
doc = builder.parse((InputStream)name);
} finally {
((InputStream)name).close();
}
} else if (name instanceof Element) {
root = (Element)name;
}

if (doc == null && root == null) {
if (quiet)
return;
throw new RuntimeException(name + " not found");
}

if (root == null) {
root = doc.getDocumentElement();
}
if (!"configuration".equals(root.getTagName()))
LOG.fatal("bad conf file: top-level element not <configuration>");
NodeList props = root.getChildNodes();
for (int i = 0; i < props.getLength(); i++) {
Node propNode = props.item(i);
if (!(propNode instanceof Element))
continue;
Element prop = (Element)propNode;
if ("configuration".equals(prop.getTagName())) {
loadResource(properties, prop, quiet);
continue;
}
if (!"property".equals(prop.getTagName()))
LOG.warn("bad conf file: element not <property>");
NodeList fields = prop.getChildNodes();
String attr = null;
String value = null;
boolean finalParameter = false;
for (int j = 0; j < fields.getLength(); j++) {
Node fieldNode = fields.item(j);
if (!(fieldNode instanceof Element))
continue;
Element field = (Element)fieldNode;
if ("name".equals(field.getTagName()) && field.hasChildNodes())
attr = ((Text)field.getFirstChild()).getData().trim();
if ("value".equals(field.getTagName()) && field.hasChildNodes())
value = ((Text)field.getFirstChild()).getData();
if ("final".equals(field.getTagName()) && field.hasChildNodes())
finalParameter = "true".equals(((Text)field.getFirstChild()).getData());
}

// Ignore this parameter if it has already been marked as 'final'
if (attr != null) {
if (value != null) {
if (!finalParameters.contains(attr)) {
properties.setProperty(attr, value);
if (storeResource) {
updatingResource.put(attr, name.toString());
}
} else if (!value.equals(properties.getProperty(attr))) {

d0cc
LOG.warn(name+":a attempt to override final parameter: "+attr
+";  Ignoring.");
}
}
if (finalParameter) {
finalParameters.add(attr);
}
}
}

} catch (IOException e) {
LOG.fatal("error parsing conf file: " + e);
throw new RuntimeException(e);
} catch (DOMException e) {
LOG.fatal("error parsing conf file: " + e);
throw new RuntimeException(e);
} catch (SAXException e) {
LOG.fatal("error parsing conf file: " + e);
throw new RuntimeException(e);
} catch (ParserConfigurationException e) {
LOG.fatal("error parsing conf file: " + e);
throw new RuntimeException(e);
}
}

使用get*和set*访问和设置配置项

get方法在Configuration对象中获取相应的配置信息。

public String get(String name, String defaultValue)

Configuration.get()会调用Conguration的私有方法substituteVars()，完成配置属性的扩展，也就是把包含${key}这种格式的变量，自动替换成对应的值。substituteVars的工作依赖于正则表达式：

varPat: $\\{[^\\}\$\u0020]+\}

如果一次属性扩展完成以后，得到的表达式里仍然包含可扩展的变量，那么，substituteVars()需要再次进行属性扩展。为了避免进入死循环，substituteVars()使用一个非常简单而有效的策略，即属性扩展只能进行一次的次数(20次，通过Configuration的静态成员变量MAX_SUBST定义)。

private String substituteVars(String expr) {
if (expr == null) {
return null;
}
Matcher match = varPat.matcher("");
String eval = expr;
for(int s=0; s<MAX_SUBST; s++) {
match.reset(eval);
if (!match.find()) {
return eval;
}
String var = match.group();
var = var.substring(2, var.length()-1); // remove ${ .. }
String val = null;
try {
val = System.getProperty(var);
} catch(SecurityException se) {
LOG.warn("Unexpected SecurityException in Configuration", se);
}
if (val == null) {
val = getRaw(var);
}
if (val == null) {
return eval; // return literal ${var}: var is unbound
}
// substitute
eval = eval.substring(0, match.start())+val+eval.substring(match.end());
}
throw new IllegalStateException("Variable substitution depth too large: "
+ MAX_SUBST + " " + expr);
}