hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From cutt...@apache.org
Subject svn commit: r406746 - in /lucene/hadoop/trunk: ./ conf/ src/java/org/apache/hadoop/dfs/ src/java/org/apache/hadoop/fs/ src/java/org/apache/hadoop/mapred/ src/test/org/apache/hadoop/fs/
Date Mon, 15 May 2006 21:26:05 GMT
Author: cutting
Date: Mon May 15 14:26:04 2006
New Revision: 406746

URL: http://svn.apache.org/viewcvs?rev=406746&view=rev
Log:
HADOOP-212.  Permit alteration of the file block size in DFS.  Contributed by Owen.

Modified:
    lucene/hadoop/trunk/CHANGES.txt
    lucene/hadoop/trunk/conf/hadoop-default.xml
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/Block.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/ClientProtocol.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSClient.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSFileInfo.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSck.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DfsPath.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DistributedFileSystem.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDataset.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDirectory.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/NameNode.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/FSDataOutputStream.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/FileSystem.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/LocalFileSystem.java
    lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/InputFormatBase.java
    lucene/hadoop/trunk/src/test/org/apache/hadoop/fs/DistributedFSCheck.java

Modified: lucene/hadoop/trunk/CHANGES.txt
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/CHANGES.txt?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/CHANGES.txt (original)
+++ lucene/hadoop/trunk/CHANGES.txt Mon May 15 14:26:04 2006
@@ -29,6 +29,12 @@
  7. HADOOP-219. Fix a NullPointerException when handling a checksum
     exception under SequenceFile.Sorter.sort().  (cutting & stack)
 
+ 8. HADOOP-212. Permit alteration of the file block size in DFS.  The
+    default block size for new files may now be specified in the
+    configuration with the dfs.block.size property.  The block size
+    may also be specified when files are opened.
+    (omalley via cutting)
+
 
 Release 0.2.1 - 2006-05-12
 

Modified: lucene/hadoop/trunk/conf/hadoop-default.xml
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/conf/hadoop-default.xml?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/conf/hadoop-default.xml (original)
+++ lucene/hadoop/trunk/conf/hadoop-default.xml Mon May 15 14:26:04 2006
@@ -135,6 +135,12 @@
 </property>
 
 <property>
+  <name>dfs.block.size</name>
+  <value>67108864</value>
+  <description>The default block size for new files.</description>
+</property>
+
+<property>
   <name>dfs.df.interval</name>
   <value>3000</value>
   <description>Disk usage statistics refresh interval in msec.</description>

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/Block.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/Block.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/Block.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/Block.java Mon May 15 14:26:04 2006
@@ -110,10 +110,9 @@
     public void readFields(DataInput in) throws IOException {
         this.blkid = in.readLong();
         this.len = in.readLong();
-        if( len < 0 || len > FSConstants.BLOCK_SIZE )
-          throw new IOException("Unexpected block size: " + len 
-                                + ". System block size = " 
-                                + FSConstants.BLOCK_SIZE + ".");
+        if( len < 0 ) {
+          throw new IOException("Unexpected block size: " + len);
+        }
     }
 
     /////////////////////////////////////

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/ClientProtocol.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/ClientProtocol.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/ClientProtocol.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/ClientProtocol.java Mon May 15 14:26:04
2006
@@ -55,7 +55,8 @@
                                 String clientName, 
                                 String clientMachine, 
                                 boolean overwrite, 
-                                short replication
+                                short replication,
+                                long blockSize
                               ) throws IOException;
 
     /**
@@ -231,4 +232,12 @@
      * One DatanodeInfo object is returned for each DataNode.
      */
     public DatanodeInfo[] getDatanodeReport() throws IOException;
+
+    /**
+     * Get the block size for the given file.
+     * @param filename The name of the file
+     * @return The number of bytes in each block
+     * @throws IOException
+     */
+    public long getBlockSize(String filename) throws IOException;
 }

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSClient.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSClient.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSClient.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSClient.java Mon May 15 14:26:04
2006
@@ -41,6 +41,7 @@
 class DFSClient implements FSConstants {
     public static final Logger LOG = LogFormatter.getLogger("org.apache.hadoop.fs.DFSClient");
     static int MAX_BLOCK_ACQUIRE_FAILURES = 3;
+    private static final long DEFAULT_BLOCK_SIZE = 64 * 1024 * 1024;
     ClientProtocol namenode;
     String localName;
     boolean running = true;
@@ -48,6 +49,8 @@
     String clientName;
     Daemon leaseChecker;
     private Configuration conf;
+    private long defaultBlockSize;
+    private short defaultReplication;
     
     // required for unknown reason to make WritableFactories work distributed
     static { new DFSFileInfo(); }
@@ -110,6 +113,8 @@
         } else {
             this.clientName = "DFSClient_" + r.nextInt();
         }
+        defaultBlockSize = conf.getLong("dfs.block.size", DEFAULT_BLOCK_SIZE);
+        defaultReplication = (short) conf.getInt("dfs.replication", 3);
         this.leaseChecker = new Daemon(new LeaseChecker());
         this.leaseChecker.start();
     }
@@ -151,6 +156,37 @@
     }
 
     /**
+     * Get the default block size for this cluster
+     * @return the default block size in bytes
+     */
+    public long getDefaultBlockSize() {
+      return defaultBlockSize;
+    }
+    
+    public long getBlockSize(Path f) throws IOException {
+      // if we already know the answer, use it.
+      if (f instanceof DfsPath) {
+        return ((DfsPath) f).getBlockSize();
+      }
+      int retries = 4;
+      while (true) {
+        try {
+          return namenode.getBlockSize(f.toString());
+        } catch (IOException ie) {
+          LOG.info("Problem getting block size: " + 
+                   StringUtils.stringifyException(ie));
+          if (--retries == 0) {
+            throw ie;
+          }
+        }
+      }
+    }
+    
+    public short getDefaultReplication() {
+      return defaultReplication;
+    }
+    
+    /**
      * Get hints about the location of the indicated block(s).  The
      * array returned is as long as there are blocks in the indicated
      * range.  Each block may have one or more locations.
@@ -182,7 +218,7 @@
     public FSOutputStream create( UTF8 src, 
                                   boolean overwrite
                                 ) throws IOException {
-      return create( src, overwrite, (short)conf.getInt("dfs.replication", 3));
+      return create( src, overwrite, defaultReplication, defaultBlockSize);
     }
     
     /**
@@ -197,10 +233,12 @@
      */
     public FSOutputStream create( UTF8 src, 
                                   boolean overwrite, 
-                                  short replication
+                                  short replication,
+                                  long blockSize
                                 ) throws IOException {
       checkOpen();
-      FSOutputStream result = new DFSOutputStream(src, overwrite, replication);
+      FSOutputStream result = new DFSOutputStream(src, overwrite, 
+                                                  replication, blockSize);
       synchronized (pendingCreates) {
         pendingCreates.put(src.toString(), result);
       }
@@ -672,15 +710,19 @@
         private long filePos = 0;
         private int bytesWrittenToBlock = 0;
         private String datanodeName;
+        private long blockSize;
 
         /**
          * Create a new output stream to the given DataNode.
          */
-        public DFSOutputStream(UTF8 src, boolean overwrite, short replication) throws IOException
{
+        public DFSOutputStream(UTF8 src, boolean overwrite, 
+                               short replication, long blockSize
+                               ) throws IOException {
             this.src = src;
             this.overwrite = overwrite;
             this.replication = replication;
             this.backupFile = newBackupFile();
+            this.blockSize = blockSize;
             this.backupStream = new FileOutputStream(backupFile);
         }
 
@@ -766,7 +808,7 @@
             while (true) {
               try {
                 return namenode.create(src.toString(), clientName.toString(),
-                    localName, overwrite, replication);
+                    localName, overwrite, replication, blockSize);
               } catch (RemoteException e) {
                 if (--retries == 0 || 
                     "org.apache.hadoop.dfs.NameNode.AlreadyBeingCreatedException".
@@ -835,7 +877,7 @@
                 throw new IOException("Stream closed");
             }
 
-            if ((bytesWrittenToBlock + pos == BLOCK_SIZE) ||
+            if ((bytesWrittenToBlock + pos == blockSize) ||
                 (pos >= BUFFER_SIZE)) {
                 flush();
             }
@@ -861,7 +903,7 @@
               len -= toWrite;
               filePos += toWrite;
 
-              if ((bytesWrittenToBlock + pos >= BLOCK_SIZE) ||
+              if ((bytesWrittenToBlock + pos >= blockSize) ||
                   (pos == BUFFER_SIZE)) {
                 flush();
               }
@@ -877,10 +919,10 @@
                 throw new IOException("Stream closed");
             }
 
-            if (bytesWrittenToBlock + pos >= BLOCK_SIZE) {
-                flushData(BLOCK_SIZE - bytesWrittenToBlock);
+            if (bytesWrittenToBlock + pos >= blockSize) {
+                flushData((int) blockSize - bytesWrittenToBlock);
             }
-            if (bytesWrittenToBlock == BLOCK_SIZE) {
+            if (bytesWrittenToBlock == blockSize) {
                 endBlock();
             }
             flushData(pos);

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSFileInfo.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSFileInfo.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSFileInfo.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSFileInfo.java Mon May 15 14:26:04
2006
@@ -40,6 +40,7 @@
     long contentsLen;
     boolean isDir;
     short blockReplication;
+    long blockSize;
 
     /**
      */
@@ -58,6 +59,7 @@
         } else 
           this.len = this.contentsLen = node.computeFileLength();
         this.blockReplication = node.getReplication();
+        blockSize = node.getBlockSize();
     }
 
     /**
@@ -102,6 +104,14 @@
       return this.blockReplication;
     }
 
+    /**
+     * Get the block size of the file.
+     * @return the number of bytes
+     */
+    public long getBlockSize() {
+      return blockSize;
+    }
+    
     //////////////////////////////////////////////////
     // Writable
     //////////////////////////////////////////////////
@@ -111,6 +121,7 @@
         out.writeLong(contentsLen);
         out.writeBoolean(isDir);
         out.writeShort(blockReplication);
+        out.writeLong(blockSize);
     }
 
     public void readFields(DataInput in) throws IOException {
@@ -120,6 +131,7 @@
         this.contentsLen = in.readLong();
         this.isDir = in.readBoolean();
         this.blockReplication = in.readShort();
+        blockSize = in.readLong();
     }
 }
 

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSck.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSck.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSck.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DFSck.java Mon May 15 14:26:04 2006
@@ -106,7 +106,7 @@
   public Result fsck(String path) throws Exception {
     DFSFileInfo[] files = dfs.listPaths(new UTF8(path));
     Result res = new Result();
-    res.setReplication(conf.getInt("dfs.replication", 3));
+    res.setReplication(dfs.getDefaultReplication());
     for (int i = 0; i < files.length; i++) {
       check(files[i], res);
     }

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DfsPath.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DfsPath.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DfsPath.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DfsPath.java Mon May 15 14:26:04 2006
@@ -18,7 +18,11 @@
 import org.apache.hadoop.fs.Path;
 
 
-/** DfsPath is a Path that's been annotated with some extra information. */
+/** 
+ * DfsPath is a Path that's been annotated with some extra information.
+ * The point of this class is to pass back the "common" metadata about 
+ * a file with the names in a directory listing to make accesses faster.
+ */
 class DfsPath extends Path {
     DFSFileInfo info;
 
@@ -41,5 +45,8 @@
     }
     public short getReplication() {
       return info.getReplication();
+    }
+    public long getBlockSize() {
+      return info.getBlockSize();
     }
 }

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DistributedFileSystem.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DistributedFileSystem.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DistributedFileSystem.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/DistributedFileSystem.java Mon May
15 14:26:04 2006
@@ -54,6 +54,18 @@
       return workingDir;
     }
     
+    public long getDefaultBlockSize() {
+      return dfs.getDefaultBlockSize();
+    }
+    
+    public long getBlockSize(Path f) throws IOException {
+      return dfs.getBlockSize(f);
+    }
+    
+    public short getDefaultReplication() {
+      return dfs.getDefaultReplication();
+    }
+    
     private Path makeAbsolute(Path f) {
       if (f.isAbsolute()) {
         return f;
@@ -78,9 +90,10 @@
       return dfs.open(getPath(f));
     }
 
-    public FSOutputStream createRaw(Path f, boolean overwrite, short replication)
+    public FSOutputStream createRaw(Path f, boolean overwrite, 
+                                    short replication, long blockSize)
       throws IOException {
-      return dfs.create(getPath(f), overwrite, replication);
+      return dfs.create(getPath(f), overwrite, replication, blockSize);
     }
 
     public boolean setReplicationRaw( Path src, 
@@ -203,10 +216,6 @@
       // FIXME: we should move the bad block(s) involved to a bad block
       // directory on their datanode, and then re-replicate the blocks, so that
       // no data is lost. a task may fail, but on retry it should succeed.
-    }
-
-    public long getBlockSize() {
-      return FSConstants.BLOCK_SIZE;
     }
 
     /** Return the total raw capacity of the filesystem, disregarding

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSConstants.java Mon May 15 14:26:04
2006
@@ -23,7 +23,6 @@
  * @author Mike Cafarella
  ************************************/
 interface FSConstants {
-    public static int BLOCK_SIZE = 32 * 1000 * 1000;
     public static int MIN_BLOCKS_FOR_WRITE = 5;
 
     public static final long WRITE_COMPLETE = 0xcafae11a;

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDataset.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDataset.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDataset.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDataset.java Mon May 15 14:26:04
2006
@@ -263,6 +263,7 @@
         if (isValidBlock(b)) {
             throw new IOException("Block " + b + " is valid, and cannot be written to.");
         }
+        long blockSize = b.getNumBytes();
 
         //
         // Serialize access to /tmp, and check if file already there.
@@ -279,7 +280,7 @@
             //
             // Check if we have too little space
             //
-            if (getRemaining() < BLOCK_SIZE) {
+            if (getRemaining() < blockSize) {
                 throw new IOException("Insufficient space for an additional block");
             }
 
@@ -288,7 +289,7 @@
             // 'reserved' size, & create file
             //
             ongoingCreates.add(b);
-            reserved += BLOCK_SIZE;
+            reserved += blockSize;
             f = getTmpFile(b);
 	    try {
 		if (f.exists()) {
@@ -304,7 +305,7 @@
 	    } catch (IOException ie) {
                 System.out.println("Exception!  " + ie);
 		ongoingCreates.remove(b);		
-		reserved -= BLOCK_SIZE;
+		reserved -= blockSize;
                 throw ie;
 	    }
         }
@@ -358,7 +359,7 @@
             if (! ongoingCreates.remove(b)) {
                 throw new IOException("Tried to finalize block " + b + ", but could not find
it in ongoingCreates after file-move!");
             } 
-            reserved -= BLOCK_SIZE;
+            reserved -= b.getNumBytes();
         }
     }
 

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDirectory.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDirectory.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDirectory.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSDirectory.java Mon May 15 14:26:04
2006
@@ -241,6 +241,18 @@
         }
 
         /**
+         * Get the block size of the first block
+         * @return the number of bytes
+         */
+        public long getBlockSize() {
+          if (blocks == null || blocks.length == 0) {
+            return 0;
+          } else {
+            return blocks[0].getNumBytes();
+          }
+        }
+        
+        /**
          */
         void listContents(Vector v) {
             if (parent != null && blocks != null) {
@@ -747,7 +759,27 @@
       }
       return fileBlocks;
     }
-                                 
+
+    /**
+     * Get the blocksize of a file
+     * @param filename the filename
+     * @return the number of bytes in the first block
+     * @throws IOException if it is a directory or does not exist.
+     */
+    public long getBlockSize(String filename) throws IOException {
+      synchronized (rootDir) {
+        INode fileNode = rootDir.getNode(filename);
+        if (fileNode == null) {
+          throw new IOException("Unknown file: " + filename);
+        }
+        if (fileNode.isDir()) {
+          throw new IOException("Getting block size of a directory: " + 
+                                filename);
+        }
+        return fileNode.getBlockSize();
+      }
+    }
+    
     /**
      * Remove the file from management, return blocks
      */

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/FSNamesystem.java Mon May 15 14:26:04
2006
@@ -37,10 +37,6 @@
  ***************************************************/
 class FSNamesystem implements FSConstants {
     public static final Logger LOG = LogFormatter.getLogger("org.apache.hadoop.fs.FSNamesystem");
-    static {
-      // for debugging the pending Creates problems
-      LOG.setLevel(Level.FINE);
-    }
 
     //
     // Stores the correct file name hierarchy
@@ -276,6 +272,10 @@
       return true;
     }
     
+    public long getBlockSize(String filename) throws IOException {
+      return dir.getBlockSize(filename);
+    }
+    
     /**
      * Check whether the replication parameter is within the range
      * determined by system configuration.
@@ -312,7 +312,8 @@
                                             UTF8 holder, 
                                             UTF8 clientMachine, 
                                             boolean overwrite,
-                                            short replication 
+                                            short replication,
+                                            long blockSize
                                           ) throws IOException {
       NameNode.stateChangeLog.fine("DIR* NameSystem.startFile: file "
             +src+" for "+holder+" at "+clientMachine);
@@ -340,7 +341,8 @@
         }
 
         // Get the array of replication targets 
-        DatanodeInfo targets[] = chooseTargets(replication, null, clientMachine);
+        DatanodeInfo targets[] = chooseTargets(replication, null, 
+                                               clientMachine, blockSize);
         if (targets.length < this.minReplication) {
             throw new IOException("failed to create file "+src
                     +" on client " + clientMachine
@@ -351,6 +353,7 @@
         // Reserve space for this pending file
         pendingCreates.put(src, 
                            new FileUnderConstruction(replication, 
+                                                     blockSize,
                                                      holder,
                                                      clientMachine));
         NameNode.stateChangeLog.finer( "DIR* NameSystem.startFile: "
@@ -421,7 +424,7 @@
         
         // Get the array of replication targets 
         DatanodeInfo targets[] = chooseTargets(pendingFile.getReplication(), 
-            null, pendingFile.getClientMachine());
+            null, pendingFile.getClientMachine(), pendingFile.getBlockSize());
         if (targets.length < this.minReplication) {
           throw new IOException("File " + src + " could only be replicated to " +
                                 targets.length + " nodes, instead of " +
@@ -1400,6 +1403,7 @@
                     }
 
                     Block block = (Block) it.next();
+                    long blockSize = block.getNumBytes();
                     FSDirectory.INode fileINode = dir.getFileByBlock(block);
                     if( fileINode == null ) { // block does not belong to any file
                         it.remove();
@@ -1411,7 +1415,13 @@
                         if (containingNodes.contains(srcNode) 
                               && ( excessBlocks == null 
                                || ! excessBlocks.contains(block))) {
-                            DatanodeInfo targets[] = chooseTargets(Math.min(fileINode.getReplication()
- containingNodes.size(), this.maxReplicationStreams - xmitsInProgress), containingNodes,
null);
+                            DatanodeInfo targets[] = 
+                              chooseTargets(Math.min(fileINode.getReplication() 
+                                                       - containingNodes.size(), 
+                                                     maxReplicationStreams
+                                                       - xmitsInProgress), 
+                                                     containingNodes, null, 
+                                                     blockSize);
                             if (targets.length > 0) {
                                 // Build items to return
                                 replicateBlocks.add(block);
@@ -1481,7 +1491,8 @@
      * considered targets.
      * @return array of DatanodeInfo instances uses as targets.
      */
-    DatanodeInfo[] chooseTargets(int desiredReplicates, TreeSet forbiddenNodes, UTF8 clientMachine)
{
+    DatanodeInfo[] chooseTargets(int desiredReplicates, TreeSet forbiddenNodes,
+                                 UTF8 clientMachine, long blockSize) {
         if (desiredReplicates > datanodeMap.size()) {
           LOG.warning("Replication requested of "+desiredReplicates
                       +" is larger than cluster size ("+datanodeMap.size()
@@ -1493,7 +1504,8 @@
         Vector targets = new Vector();
 
         for (int i = 0; i < desiredReplicates; i++) {
-            DatanodeInfo target = chooseTarget(forbiddenNodes, alreadyChosen, clientMachine);
+            DatanodeInfo target = chooseTarget(forbiddenNodes, alreadyChosen, 
+                                               clientMachine, blockSize);
             if (target == null)
               break; // calling chooseTarget again won't help
             targets.add(target);
@@ -1514,7 +1526,8 @@
      * @return DatanodeInfo instance to use or null if something went wrong
      * (a log message is emitted if null is returned).
      */
-    DatanodeInfo chooseTarget(TreeSet forbidden1, TreeSet forbidden2, UTF8 clientMachine)
{
+    DatanodeInfo chooseTarget(TreeSet forbidden1, TreeSet forbidden2, 
+                              UTF8 clientMachine, long blockSize) {
         //
         // Check if there are any available targets at all
         //
@@ -1565,7 +1578,7 @@
                 for (Iterator it = targetList.iterator(); it.hasNext(); ) {
                     DatanodeInfo node = (DatanodeInfo) it.next();
                     if (clientMachine.equals(node.getHost())) {
-                        if (node.getRemaining() > BLOCK_SIZE * MIN_BLOCKS_FOR_WRITE) {
+                        if (node.getRemaining() > blockSize * MIN_BLOCKS_FOR_WRITE) {
                             return node;
                         }
                     }
@@ -1577,7 +1590,7 @@
             //
             for (Iterator it = targetList.iterator(); it.hasNext(); ) {
                 DatanodeInfo node = (DatanodeInfo) it.next();
-                if (node.getRemaining() > BLOCK_SIZE * MIN_BLOCKS_FOR_WRITE) {
+                if (node.getRemaining() > blockSize * MIN_BLOCKS_FOR_WRITE) {
                     return node;
                 }
             }
@@ -1589,7 +1602,7 @@
             //
             for (Iterator it = targetList.iterator(); it.hasNext(); ) {
                 DatanodeInfo node = (DatanodeInfo) it.next();
-                if (node.getRemaining() > BLOCK_SIZE) {
+                if (node.getRemaining() > blockSize) {
                     return node;
                 }
             }
@@ -1615,14 +1628,17 @@
      */
     private class FileUnderConstruction {
       private short blockReplication; // file replication
+      private long blockSize;
       private Vector blocks;
       private UTF8 clientName;         // lease holder
       private UTF8 clientMachine;
       
       FileUnderConstruction(short replication,
+                            long blockSize,
                             UTF8 clientName,
                             UTF8 clientMachine) throws IOException {
         this.blockReplication = replication;
+        this.blockSize = blockSize;
         this.blocks = new Vector();
         this.clientName = clientName;
         this.clientMachine = clientMachine;
@@ -1630,6 +1646,10 @@
       
       public short getReplication() {
         return this.blockReplication;
+      }
+      
+      public long getBlockSize() {
+        return blockSize;
       }
       
       public Vector getBlocks() {

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/NameNode.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/NameNode.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/NameNode.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/dfs/NameNode.java Mon May 15 14:26:04 2006
@@ -181,7 +181,8 @@
                                String clientName, 
                                String clientMachine, 
                                boolean overwrite,
-                               short replication
+                               short replication,
+                               long blockSize
     ) throws IOException {
        stateChangeLog.fine("*DIR* NameNode.create: file "
             +src+" for "+clientName+" at "+clientMachine);
@@ -189,7 +190,8 @@
                                                 new UTF8(clientName), 
                                                 new UTF8(clientMachine), 
                                                 overwrite,
-                                                replication);
+                                                replication,
+                                                blockSize);
         Block b = (Block) results[0];
         DatanodeInfo targets[] = (DatanodeInfo[]) results[1];
         return new LocatedBlock(b, targets);
@@ -279,6 +281,11 @@
             return results;
         }
     }
+    
+    public long getBlockSize(String filename) throws IOException {
+      return namesystem.getBlockSize(filename);
+    }
+    
     /**
      */
     public boolean rename(String src, String dst) throws IOException {

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/FSDataOutputStream.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/FSDataOutputStream.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/FSDataOutputStream.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/FSDataOutputStream.java Mon May 15 14:26:04
2006
@@ -38,12 +38,14 @@
                   Path file, 
                   boolean overwrite, 
                   short replication,
+                  long blockSize,
                   Configuration conf)
       throws IOException {
-      super(fs.createRaw(file, overwrite, replication));
+      super(fs.createRaw(file, overwrite, replication, blockSize));
       this.bytesPerSum = conf.getInt("io.bytes.per.checksum", 512);
       this.sums = new FSDataOutputStream(
-            fs.createRaw(FileSystem.getChecksumFile(file), true, replication), 
+            fs.createRaw(FileSystem.getChecksumFile(file), true, 
+                         replication, blockSize), 
             conf);
       sums.write(CHECKSUM_VERSION, 0, CHECKSUM_VERSION.length);
       sums.writeInt(this.bytesPerSum);
@@ -127,11 +129,11 @@
 
   public FSDataOutputStream(FileSystem fs, Path file,
                             boolean overwrite, Configuration conf,
-                            int bufferSize, short replication )
+                            int bufferSize, short replication, long blockSize )
   throws IOException {
     super(new Buffer(
             new PositionCache(
-                new Summer(fs, file, overwrite, replication, conf)), 
+                new Summer(fs, file, overwrite, replication, blockSize, conf)), 
             bufferSize));
   }
 

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/FileSystem.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/FileSystem.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/FileSystem.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/FileSystem.java Mon May 15 14:26:04
2006
@@ -175,7 +175,8 @@
     public FSDataOutputStream create(Path f) throws IOException {
       return create(f, true, 
                     getConf().getInt("io.file.buffer.size", 4096),
-                    (short)getConf().getInt("dfs.replication", 3));
+                    getDefaultReplication(),
+                    getDefaultBlockSize());
     }
 
     /**
@@ -186,7 +187,8 @@
       throws IOException {
       return create(f, true, 
                     getConf().getInt("io.file.buffer.size", 4096),
-                    replication);
+                    replication,
+                    getDefaultBlockSize());
     }
 
     /**
@@ -201,7 +203,8 @@
                                       int bufferSize
                                     ) throws IOException {
       return create( f, overwrite, bufferSize, 
-                    (short)getConf().getInt("dfs.replication", 3));
+                     getDefaultReplication(),
+                     getDefaultBlockSize());
     }
     
     /**
@@ -215,10 +218,11 @@
     public FSDataOutputStream create( Path f, 
                                       boolean overwrite,
                                       int bufferSize,
-                                      short replication
+                                      short replication,
+                                      long blockSize
                                     ) throws IOException {
       return new FSDataOutputStream(this, f, overwrite, getConf(), 
-                                    bufferSize, replication );
+                                    bufferSize, replication, blockSize );
     }
 
     /** Opens an OutputStream at the indicated Path.
@@ -227,7 +231,9 @@
      *   the file will be overwritten, and if false an error will be thrown.
      * @param replication required block replication for the file. 
      */
-    public abstract FSOutputStream createRaw(Path f, boolean overwrite, short replication)
+    public abstract FSOutputStream createRaw(Path f, boolean overwrite, 
+                                             short replication,
+                                             long blockSize)
       throws IOException;
 
     /** @deprecated Call {@link #createNewFile(Path)} instead. */
@@ -547,8 +553,20 @@
                                                long start, long length,
                                                int crc);
 
+    /**
+     * Get the size for a particular file.
+     * @param f the filename
+     * @return the number of bytes in a block
+     */
+    public abstract long getBlockSize(Path f) throws IOException;
+    
     /** Return the number of bytes that large input files should be optimally
      * be split into to minimize i/o time. */
-    public abstract long getBlockSize();
+    public abstract long getDefaultBlockSize();
+    
+    /**
+     * Get the default replication.
+     */
+    public abstract short getDefaultReplication();
 
 }

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/LocalFileSystem.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/LocalFileSystem.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/LocalFileSystem.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/fs/LocalFileSystem.java Mon May 15 14:26:04
2006
@@ -161,7 +161,8 @@
       }
     }
 
-    public FSOutputStream createRaw(Path f, boolean overwrite, short replication)
+    public FSOutputStream createRaw(Path f, boolean overwrite, 
+                                    short replication, long blockSize)
       throws IOException {
         if (exists(f) && ! overwrite) {
             throw new IOException("File already exists:"+f);
@@ -356,9 +357,18 @@
       }
     }
 
-    public long getBlockSize() {
+    public long getDefaultBlockSize() {
       // default to 32MB: large enough to minimize the impact of seeks
       return getConf().getLong("fs.local.block.size", 32 * 1024 * 1024);
+    }
+
+    public long getBlockSize(Path filename) {
+      // local doesn't really do blocks, so just use the global number
+      return getDefaultBlockSize();
+    }
+    
+    public short getDefaultReplication() {
+      return 1;
     }
 
 

Modified: lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/InputFormatBase.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/InputFormatBase.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/InputFormatBase.java (original)
+++ lucene/hadoop/trunk/src/java/org/apache/hadoop/mapred/InputFormatBase.java Mon May 15
14:26:04 2006
@@ -32,7 +32,7 @@
   public static final Logger LOG =
     LogFormatter.getLogger("org.apache.hadoop.mapred.InputFormatBase");
 
-  private static final double SPLIT_SLOP = 0.1;   // 10% slop
+  private static final double SPLIT_SLOP = 1.1;   // 10% slop
 
   private long minSplitSize = 1;
 
@@ -117,34 +117,22 @@
       totalSize += fs.getLength(files[i]);
     }
 
-    long bytesPerSplit = totalSize / numSplits;   // start w/ desired num splits
+    long goalSize = totalSize / numSplits;   // start w/ desired num splits
 
-    long fsBlockSize = fs.getBlockSize();
-    if (bytesPerSplit > fsBlockSize) {            // no larger than fs blocks
-      bytesPerSplit = fsBlockSize;
-    }
-
-    long configuredMinSplitSize = job.getLong("mapred.min.split.size", 0);
-    if( configuredMinSplitSize < minSplitSize )
-    	configuredMinSplitSize = minSplitSize;
-    if (bytesPerSplit < configuredMinSplitSize) { // no smaller than min size
-      bytesPerSplit = configuredMinSplitSize;
-    }
-
-    long maxPerSplit = bytesPerSplit + (long)(bytesPerSplit*SPLIT_SLOP);
-
-    //LOG.info("bytesPerSplit = " + bytesPerSplit);
-    //LOG.info("maxPerSplit = " + maxPerSplit);
+    long minSize = Math.max(job.getLong("mapred.min.split.size", 1),
+                            minSplitSize);
 
     ArrayList splits = new ArrayList(numSplits);  // generate splits
     for (int i = 0; i < files.length; i++) {
       Path file = files[i];
       long length = fs.getLength(file);
+      long blockSize = fs.getBlockSize(file);
+      long splitSize = computeSplitSize(goalSize, minSize, blockSize);
 
       long bytesRemaining = length;
-      while (bytesRemaining >= maxPerSplit) {
-        splits.add(new FileSplit(file, length-bytesRemaining, bytesPerSplit));
-        bytesRemaining -= bytesPerSplit;
+      while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
+        splits.add(new FileSplit(file, length-bytesRemaining, splitSize));
+        bytesRemaining -= splitSize;
       }
       
       if (bytesRemaining != 0) {
@@ -156,5 +144,9 @@
     return (FileSplit[])splits.toArray(new FileSplit[splits.size()]);
   }
 
+  private static long computeSplitSize(long goalSize, long minSize,
+                                       long blockSize) {
+    return Math.max(minSize, Math.min(goalSize, blockSize));
+  }
 }
 

Modified: lucene/hadoop/trunk/src/test/org/apache/hadoop/fs/DistributedFSCheck.java
URL: http://svn.apache.org/viewcvs/lucene/hadoop/trunk/src/test/org/apache/hadoop/fs/DistributedFSCheck.java?rev=406746&r1=406745&r2=406746&view=diff
==============================================================================
--- lucene/hadoop/trunk/src/test/org/apache/hadoop/fs/DistributedFSCheck.java (original)
+++ lucene/hadoop/trunk/src/test/org/apache/hadoop/fs/DistributedFSCheck.java Mon May 15 14:26:04
2006
@@ -104,7 +104,7 @@
     if( ! fs.isDirectory(rootFile) ) {
       nrFiles++;
       // For a regular file generate <fName,offset> pairs
-      long blockSize = fs.getBlockSize();
+      long blockSize = fs.getDefaultBlockSize();
       long fileLength = fs.getLength( rootFile );
       for( long offset = 0; offset < fileLength; offset += blockSize )
         writer.append(new UTF8(rootFile.toString()), new LongWritable(offset));
@@ -136,7 +136,7 @@
       in = new DataInputStream(fs.open(new Path(name)));
       long actualSize = 0;
       try {
-        long blockSize = fs.getBlockSize();
+        long blockSize = fs.getDefaultBlockSize();
         int curSize = bufferSize;
         for(  actualSize = 0; 
               curSize == bufferSize && actualSize < blockSize;



Mime
View raw message