public interface FsVolumeSpi {
//获取volume下的存储UUID
public String getStorageID();
//获取BlockPool的列表
public String[] getBlockPoolList();
//获取可用的存储空间大小
public long getAvailable() throws IOException;
//获取volume的基本路径(current目录的父目录路径)
public String getBasePath();
//获取volume的绝对路径
public String getPath(String bpid) throws IOException;
//获取block的提交目录
public File getFinalizedDir(String bpid) throws IOException;
//获取存储类型
public StorageType getStorageType();
} RollingLogs接口用于对DataNode上的EditLog进行管理，内部包含两个重要的接口。LineIterator用于遍历EditLog的内容，Appender接口用于追加内容到EditLog。 VolumeChoosingPolicy接口提供写入block副本的volume选择策略，实现类有两个，和接口在同一个包下,分别是RoundRobinVolumeChoosingPolicy和AvailableSpaceVolumeChoosingPolicy。 org.apache.hadoop.hdfs.server.datanode.fsdataset.impl主要包括相关接口的实现类和一些工具类。FsDataSet将存储空间分为三个级别，LDir(cdh3x的FSDir),FsVolumeImpl(cdh3x的FsVolume)和FsVolumeList(cdh3x的FsVolumeSet)。LDir代表current目录下的子目录，FsVolumeImpl代表${dfs.datanode.data.dir}中的一项，DataNode可以有多个数据目录，FsVolumeList负责管理FsVolumeImpl对象。 以创建数据块为入口，详细分析下代码的执行流程，在HDFS支持Append特性之前，一个block副本在DataNode的状态要么是finalized，要么是temporary。temporary状态的block会在DataNode重启过程中被删除。但支持了Append之后，HDFS必须为正在构建中的block提供更强的可持久化支持，一些temporary的block副本必须在DataNode重启过程中持久存在。所以在后续的版本中代码的变动比较大。详情参考https://issues.apache.org/jira/browse/HDFS-265。 在cdh3x中，FsDataSet中的writeToBlock方法用于选择block写入的volume，并创建写入的输出流，新版本中将这个方法拆分为三个方法，分别是createRbw，createTemporary和append。详情见https://issues.apache.org/jira/browse/HDFS-543。rbw状态代表一个block副本刚刚被创建或者被Append，temporary则代表DataNode间block的备份和reblance。下面详细分析下客户端写入文件创建block输出流的过程。首先看一段cdh3x的代码以做对比BlockReceiver(Block block, DataInputStream in, String inAddr,
String myAddr, boolean isRecovery, String clientName,
DatanodeInfo srcDataNode, DataNode datanode) throws IOException {
//通过调用writeToBlock方法创建流对象
streams = datanode.data.writeToBlock(block, isRecovery,
clientName == null || clientName.length() == 0);
this.finalized = false;
if (streams != null) {
//获取输出流对象
this.out = streams.dataOut;
//获取校验信息输出流对象
this.checksumOut = new DataOutputStream(new BufferedOutputStream(streams.checksumOut, SMALL_BUFFER_SIZE));
}
} cdh3x中通过调用writeToBlock来创建对应的输出流，而在新代码中获取输出流对象的逻辑相对复杂一些。代码如下 //如果是DataNode节点发起的数据块复制或者移动
if (isDatanode) {
//创建temporary状态的replica
replicaInfo = datanode.data.createTemporary(block);
} else {
//如果是客户端发起的,根据所处的阶段执行不同的操作
switch (stage) {
//pipeline启动创建
case PIPELINE_SETUP_CREATE:
//创建rbw状态的replicaInfo
replicaInfo = datanode.data.createRbw(block);
//向NameNode已经创建block replica的消息
datanode.notifyNamenodeReceivingBlock(block, replicaInfo.getStorageUuid());
break;
case PIPELINE_SETUP_STREAMING_RECOVERY:
//获取rbw恢复的replicaInfo
replicaInfo = datanode.data.recoverRbw(block, newGs, minBytesRcvd, maxBytesRcvd);
//设置block新版本号
block.setGenerationStamp(newGs);
break;
case PIPELINE_SETUP_APPEND:
//block追加操作
replicaInfo = datanode.data.append(block, newGs, minBytesRcvd);
if (datanode.blockScanner != null) {
//通过blockScanner删除旧的block
datanode.blockScanner.deleteBlock(block.getBlockPoolId(),block.getLocalBlock());
}
//设置block的新版本号
block.setGenerationStamp(newGs);
//向NameNode发送变更消息
datanode.notifyNamenodeReceivingBlock(block, replicaInfo.getStorageUuid());
break;
case PIPELINE_SETUP_APPEND_RECOVERY:
//获取append恢复的replicaInfo
replicaInfo = datanode.data.recoverAppend(block, newGs, minBytesRcvd);
if (datanode.blockScanner != null) {
//通过blockScanner删除旧的block
datanode.blockScanner.deleteBlock(block.getBlockPoolId(),
block.getLocalBlock());
}
//设置block的新版本号
block.setGenerationStamp(newGs);
//向NameNode发送变更消息
datanode.notifyNamenodeReceivingBlock(block, replicaInfo.getStorageUuid());
break;
case TRANSFER_RBW:
case TRANSFER_FINALIZED:
//DataNode之间传输创建的replicaInfo
replicaInfo = datanode.data.createTemporary(block);
break;
default: throw new IOException("Unsupported stage " + stage +
" while receiving block " + block + " from " + inAddr);
}
}
this.dropCacheBehindWrites = (cachingStrategy.getDropBehind() == null) ?
datanode.getDnConf().dropCacheBehindWrites :
cachingStrategy.getDropBehind();
this.syncBehindWrites = datanode.getDnConf().syncBehindWrites;
final boolean isCreate = isDatanode || isTransfer
|| stage == BlockConstructionStage.PIPELINE_SETUP_CREATE;
//创建输出流对象
streams = replicaInfo.createStreams(isCreate, requestedChecksum);
//获取校验相关的信息
this.clientChecksum = requestedChecksum;
this.diskChecksum = streams.getChecksum();
this.needsChecksumTranslation = !clientChecksum.equals(diskChecksum);
this.bytesPerChecksum = diskChecksum.getBytesPerChecksum();
this.checksumSize = diskChecksum.getChecksumSize();
//获取数据输出流
this.out = streams.getDataOut(); 由于支持了Append操作，block副本所处的状态更多了，代码实现上也更加复杂。下面详细分析下FsDatasetImpl类下的createRbw方法public synchronized ReplicaInPipeline createRbw(ExtendedBlock b)
throws IOException {
//从volumeMap中获取replicaInfo,因为是新创建的，如果在volumeMap存在，则不能继续创建
ReplicaInfo replicaInfo = volumeMap.get(b.getBlockPoolId(),
b.getBlockId());
if (replicaInfo != null) {
throw new ReplicaAlreadyExistsException("Block " + b +
" already exists in state " + replicaInfo.getState() +
" and thus cannot be created.");
}
//根据相应的规则获取要写入的volume
FsVolumeImpl v = volumes.getNextVolume(b.getNumBytes());
//在得到的volume下创建rbw文件
File f = v.createRbwFile(b.getBlockPoolId(), b.getLocalBlock());
//创建ReplicaBeingWritten对象,此对象继承自ReplicaInfo
ReplicaBeingWritten newReplicaInfo = new ReplicaBeingWritten(b.getBlockId(),
b.getGenerationStamp(), v, f.getParentFile());
//将新创建的replicaInfo加入到volumeMap中
volumeMap.add(b.getBlockPoolId(), newReplicaInfo);
return newReplicaInfo;
} 在选择写入的volume的时候，用户可以根据dfs.datanode.fsdataset.volume.choosing.policy属性来选择使用何种策略来选择volume。默认提供两种策略，分别是根据volume可用空间和轮询的方法来选择volume,对应的实现类分别是AvailableSpaceVolumeChoosingPolicy和RoundRobinVolumeChoosingPolicy，它们都继承自VolumeChoosingPolicy接口，用户也可以根据自己的需求来自定义选择策略。我有一个Ruby程序，它使用rubyzip压缩XML文件的目录树。gem。我的问题是文件开始变得很重，我想提高压缩级别，因为压缩时间不是问题。我在rubyzipdocumentation中找不到一种为创建的ZIP文件指定压缩级别的方法。有人知道如何更改此设置吗？是否有另一个允许指定压缩级别的Ruby库？ 最佳答案 这是我通过查看rubyzip内部创建的代码。level=Zlib::BEST_COMPRESSIONZip::ZipOutputStream.open(zip_file)do|zip|Dir.glob("**/*")d
我试图在一个项目中使用rake，如果我把所有东西都放到Rakefile中，它会很大并且很难读取/找到东西，所以我试着将每个命名空间放在lib/rake中它自己的文件中，我添加了这个到我的rake文件的顶部:Dir['#{File.dirname(__FILE__)}/lib/rake/*.rake'].map{|f|requiref}它加载文件没问题，但没有任务。我现在只有一个.rake文件作为测试，名为“servers.rake”，它看起来像这样:namespace:serverdotask:testdoputs"test"endend所以当我运行rakeserver:testid时
我的目标是转换表单输入，例如“100兆字节”或“1GB”，并将其转换为我可以存储在数据库中的文件大小(以千字节为单位)。目前，我有这个:defquota_convert@regex=/([0-9]+)(.*)s/@sizes=%w{kilobytemegabytegigabyte}m=self.quota.match(@regex)if@sizes.include?m[2]eval("self.quota=#{m[1]}.#{m[2]}")endend这有效，但前提是输入是倍数(“gigabytes”，而不是“gigabyte”)并且由于使用了eval看起来疯狂不安全。所以，功能正常，
Rails2.3可以选择随时使用RouteSet#add_configuration_file添加更多路由。是否可以在Rails3项目中做同样的事情？ 最佳答案 在config/application.rb中:config.paths.config.routes在Rails3.2(也可能是Rails3.1)中，使用:config.paths["config/routes"] 关于ruby-on-rails-Rails3中的多个路由文件，我们在StackOverflow上找到一个类似的问题
对于具有离线功能的智能手机应用程序，我正在为Xml文件创建单向文本同步。我希望我的服务器将增量/差异(例如GNU差异补丁)发送到目标设备。这是计划:Time=0Server:hasversion_1ofXmlfile(~800kiB)Client:hasversion_1ofXmlfile(~800kiB)Time=1Server:hasversion_1andversion_2ofXmlfile(each~800kiB)computesdeltaoftheseversions(=patch)(~10kiB)sendspatchtoClient(~10kiBtransferred)Cl
我正在寻找执行以下操作的正确语法(在Perl、Shell或Ruby中):#variabletoaccessthedatalinesappendedasafileEND_OF_SCRIPT_MARKERrawdatastartshereanditcontinues. 最佳答案 Perl用__DATA__做这个:#!/usr/bin/perlusestrict;usewarnings;while(){print;}__DATA__Texttoprintgoeshere 关于ruby-如何将脚
使用带有Rails插件的vim，您可以创建一个迁移文件，然后一次性打开该文件吗？textmate也可以这样吗？ 最佳答案 你可以使用rails.vim然后做类似的事情::Rgeneratemigratonadd_foo_to_bar插件将打开迁移生成的文件，这正是您想要的。我不能代表textmate。 关于ruby-使用VimRails，您可以创建一个新的迁移文件并一次性打开它吗？，我们在StackOverflow上找到一个类似的问题： https://sta
好的，所以我的目标是轻松地将一些数据保存到磁盘以备后用。您如何简单地写入然后读取一个对象？所以如果我有一个简单的类classCattr_accessor:a,:bdefinitialize(a,b)@a,@b=a,bendend所以如果我从中非常快地制作一个objobj=C.new("foo","bar")#justgaveitsomerandomvalues然后我可以把它变成一个kindaidstring=obj.to_s#whichreturns""我终于可以将此字符串打印到文件或其他内容中。我的问题是，我该如何再次将这个id变回一个对象？我知道我可以自己挑选信息并制作一个接受该信
我正在编写一个小脚本来定位aws存储桶中的特定文件，并创建一个临时验证的url以发送给同事。(理想情况下，这将创建类似于在控制台上右键单击存储桶中的文件并复制链接地址的结果)。我研究过回形针，它似乎不符合这个标准，但我可能只是不知道它的全部功能。我尝试了以下方法:defauthenticated_url(file_name,bucket)AWS::S3::S3Object.url_for(file_name,bucket,:secure=>true,:expires=>20*60)end产生这种类型的结果:...-1.amazonaws.com/file_path/file.zip.A
我注意到像bundler这样的项目在每个specfile中执行requirespec_helper我还注意到rspec使用选项--require，它允许您在引导rspec时要求一个文件。您还可以将其添加到.rspec文件中，因此只要您运行不带参数的rspec就会添加它。使用上述方法有什么缺点可以解释为什么像bundler这样的项目选择在每个规范文件中都需要spec_helper吗？ 最佳答案 我不在Bundler上工作，所以我不能直接谈论他们的做法。并非所有项目都checkin.rspec文件。原因是这个文件，通常按照当前的惯例，只