ハタさんのブログ(復刻版)

hadoop の話題。その2

hadoop を支える HDFS には HDFS-APIを通すことで、プログラム中から HDFS の読み書きが出きるようになります。(たぶん、hdfs-s3 なんかもこのAPI経由(? ソース読んでない))

(中略)

んで、この HDFS-API のなかに、Thrift を使って リモート上から HDFS の読み書きをできるようにしている HadoopThriftServer(theiftfs) があります。

この thriftfs の起動は に書かれているのですが、shellを握ってしまうのでこんな感じにしました。

#!/usr/bin/env bash

THRIFTFS_PID_FILE=$HADOOP_PID_DIR/thrift.pid
THRIFTFS_LOG_FILE=$HADOOP_LOG_DIR/thrift.log

if [ -f $THRIFTFS_PID_FILE ]; then
    echo running as process `cat $THRIFTFS_PID_FILE`. stop it first
    exit 1
fi

CLASSPATH=$HADOOP_CONF_DIR
CLASSPATH=$CLASSPATH:$HADOOP_HOME/hadoop-0.20.1-core.jar:$HADOOP_HOME/hadoop-0.20.1-tools.jar
CLASSPATH=$CLASSPATH:$HADOOP_HOME/lib/commons-logging-1.0.4.jar
CLASSPATH=$CLASSPATH:$HADOOP_HOME/lib/commons-logging-api-1.0.4.jar
CLASSPATH=$CLASSPATH:$HADOOP_HOME/lib/log4j-1.2.15.jar
CLASSPATH=$CLASSPATH:$HADOOP_HOME/contrib/thriftfs/hadoop-0.20.1-thriftfs.jar
CLASSPATH=$CLASSPATH:$HADOOP_HOME/src/contrib/thriftfs/lib/hadoopthriftapi.jar
CLASSPATH=$CLASSPATH:$HADOOP_HOME/src/contrib/thriftfs/lib/libthrift.jar

nohup java -Dcom.sun.management.jmxremote -cp $CLASSPATH org.apache.hadoop.thriftfs.HadoopThriftServer 10010 > $THRIFTFS_LOG_FILE 2>&1 < /dev/null &
echo $! > $THRIFTFS_PID_FILE

$HADOOP_HOME/bin/hadoop dfsadmin -safemode leave

同様に 停止は

#!/usr/bin/env bash

THRIFT_FS_PID_FILE=$HADOOP_PID_DIR/thrift.pid
THRIFT_FS_LOG_FILE=$HADOOP_LOG_DIR/thrift.log

if [ -f $THRIFT_FS_PID_FILE ]; then
    kill `cat $THRIFT_FS_PID_FILE`
    rm $THRIFT_FS_PID_FILE
else
    echo no pidfile $THRIFT_FS_PID_FILE
fi

起動する際は dfs($HADOOP_HOME/bin/start-dfs.sh)が起動している状態で起動する必要があります。
(dfsadmin -safemode leave は適宜行ってください)

んで、このThriftFSは、こんな感じでリモート上の DFS のファイルを読み書きできます

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hdfs.protocol.FSConstants;
import org.apache.hadoop.fs.permission.FsPermission;

import org.apache.hadoop.thriftfs.api.Pathname;
import org.apache.hadoop.thriftfs.api.ThriftHadoopFileSystem;
import org.apache.hadoop.thriftfs.api.ThriftHandle;
import org.apache.hadoop.thriftfs.api.ThriftIOException;

import com.facebook.thrift.TException;
import com.facebook.thrift.protocol.TBinaryProtocol;
import com.facebook.thrift.protocol.TProtocol;
import com.facebook.thrift.transport.TSocket;
import com.facebook.thrift.transport.TTransportException;

public class HDFSInput {
    
    public static void main(String...args) {
        final int defaultBufferSize = config.getInt("io.file.buffer.size", 4096);
        final long defaultBlockSize = config.getLong("dfs.block.size", FSConstants.DEFAULT_BLOCK_SIZE);
        final short defaultReplication = (short) config.getInt("dfs.replication", 3);

        TSocket socket = new TSocket("master1", 10010);
        TProtocol protocol = new TBinaryProtocol(socket);
        
        try {
            socket.open();
            try {
                ThriftHadoopFileSystem.Client client = new ThriftHadoopFileSystem.Client(protocol);
                // client timeout 5 sec
                client.setTimeout(5);
                Pathname hoge = new Pathname("/tmp/hoge");

                FsPermission permission = FsPermission.createImmutable((short) 0655);
                boolean overwrite = true;

                ThriftHandle writeHandler = client.createFile(hoge, permission.toShort(), overwrite, defaultBufferSize, defaultReplication, defaultBlockSize);
                client.write(writeHandler, "hello world");
                client.close(writeHandler);

                ThriftHandle readHandler = client.open(hoge);
                System.out.println(client.read(readHandler, 0, 1024));
                client.close(readHandler);
            } catch (TTransportException e) {
                e.printStackTrace();
            } catch (ThriftIOException e) {
                e.printStackTrace();
            }
        } catch (TException e) {
            e.printStackTrace();
        } finally {
            socket.close();
        }
    }
}

read時にBufferedReaderにwrapするともう少し便利に読み書きできる(これは今度書く)

んで、読み書きできるようになったんだけど、どうも複数のクライアントから連続して読み書きをすると、整合性が取れなくなってしまう(?)のかエラーがでるようになった。
元のソースを読むと...なんともエレガントな。。。

ということで、java.util.concurrent.atomic を使って書き直してみた(ここが本題)

package org.apache.hadoop.thriftfs;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.util.LinkedList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.thriftfs.api.Pathname;
import org.apache.hadoop.thriftfs.api.ThriftHadoopFileSystem;
import org.apache.hadoop.thriftfs.api.ThriftHandle;
import org.apache.hadoop.thriftfs.api.ThriftIOException;
import org.apache.hadoop.util.Daemon;
import org.apache.hadoop.util.StringUtils;

import com.facebook.thrift.protocol.TBinaryProtocol;
import com.facebook.thrift.server.TServer;
import com.facebook.thrift.server.TThreadPoolServer;
import com.facebook.thrift.transport.TServerSocket;
import com.facebook.thrift.transport.TServerTransport;
import com.facebook.thrift.transport.TTransportFactory;

public class HadoopThriftServer extends ThriftHadoopFileSystem {

    static int serverPort = 0;                    // default port
    TServer    server = null;

    public static class HadoopThriftHandler implements ThriftHadoopFileSystem.Iface
    {

      public static final Log LOG = LogFactory.getLog("org.apache.hadoop.thrift");

      // HDFS glue
      Configuration conf;
      FileSystem fs;
          
      // stucture that maps each Thrift object into an hadoop object
      private AtomicLong nextId = new AtomicLong(new Random().nextLong());
      private ConcurrentHashMap<Long, Object> hadoopHash = new ConcurrentHashMap<Long, Object>();
      private Daemon inactivityThread = null;

      // Detect inactive session
      private static volatile long inactivityPeriod = 3600 * 1000; // 1 hr
      private static volatile long inactivityRecheckInterval = 60 * 1000;
      private static volatile boolean fsRunning = true;
      private AtomicLong now = new AtomicLong(now());

      // allow outsider to change the hadoopthrift path
      public void setOption(String key, String val) {
      }

      /**
       * Current system time.
       * @return current time in msec.
       */
      static long now() {
        return System.currentTimeMillis();
      }

      /**
      * getVersion
      *
      * @return current version of the interface.
      */
      public String getVersion() {
        return "0.1";
      }

      /**
       * shutdown
       *
       * cleanly closes everything and exit.
       */
      public void shutdown(int status) {
        LOG.info("HadoopThriftServer shutting down.");
        try {
          fs.close();
        } catch (IOException e) {
          LOG.warn("Unable to close file system");
        }
        Runtime.getRuntime().exit(status);
      }

      /**
       * Periodically checks to see if there is inactivity
       */
      class InactivityMonitor implements Runnable {
        public void run() {
          while (fsRunning) {
            try {
              if (now() > now.get() + inactivityPeriod) {
                LOG.warn("HadoopThriftServer Inactivity period of " +
                         inactivityPeriod + " expired... Stopping Server.");
                shutdown(-1);
              }
            } catch (Exception e) {
              LOG.error(StringUtils.stringifyException(e));
            }
            try {
              Thread.sleep(inactivityRecheckInterval);
            } catch (InterruptedException ie) {
            }
          }
        }
      }

      /**
       * HadoopThriftServer
       *
       * Constructor for the HadoopThriftServer glue with Thrift Class.
       *
       * @param name - the name of this handler
       */
      public HadoopThriftHandler(String name) {
        conf = new Configuration();
        now.set(now());
        try {
          inactivityThread = new Daemon(new InactivityMonitor());
          fs = FileSystem.get(conf);
        } catch (IOException e) {
          LOG.warn("Unable to open hadoop file system...");
          Runtime.getRuntime().exit(-1);
        }
      }

      /**
        * printStackTrace
        *
        * Helper function to print an exception stack trace to the log and not stderr
        *
        * @param e the exception
        *
        */
      static private void printStackTrace(Exception e) {
        for(StackTraceElement s: e.getStackTrace()) {
          LOG.error(s);
        }
      }

      /**
       * Lookup a thrift object into a hadoop object
       */
      private synchronized Object lookup(long id) {
        return hadoopHash.get(new Long(id));
      }

      /**
       * Insert a thrift object into a hadoop object. Return its id.
       */
      private synchronized long insert(Object o) {
        long next = nextId.incrementAndGet();
        hadoopHash.put(next, o);
        return next;
      }

      /**
       * Delete a thrift object from the hadoop store.
       */
      private synchronized Object remove(long id) {
        return hadoopHash.remove(new Long(id));
      }

      /**
        * Implement the API exported by this thrift server
        */

      /** Set inactivity timeout period. The period is specified in seconds.
        * if there are no RPC calls to the HadoopThrift server for this much
        * time, then the server kills itself.
        */
      public synchronized void setInactivityTimeoutPeriod(long periodInSeconds) {
        inactivityPeriod = periodInSeconds * 1000; // in milli seconds
        if (inactivityRecheckInterval > inactivityPeriod ) {
          inactivityRecheckInterval = inactivityPeriod;
        }
      }


      /**
        * Create a file and open it for writing
        */
      public ThriftHandle create(Pathname path) throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("create: " + path);
          FSDataOutputStream out = fs.create(new Path(path.pathname));
          long id = insert(out);
          ThriftHandle obj = new ThriftHandle(id);
          HadoopThriftHandler.LOG.debug("created: " + path + " id: " + id);
          return obj;
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
        * Create a file and open it for writing, delete file if it exists
        */
      public ThriftHandle createFile(Pathname path, 
                                     short mode,
                                     boolean  overwrite,
                                     int bufferSize,
                                     short replication,
                                     long blockSize) throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("create: " + path +
                                       " permission: " + mode +
                                       " overwrite: " + overwrite +
                                       " bufferSize: " + bufferSize +
                                       " replication: " + replication +
                                       " blockSize: " + blockSize);
          FSDataOutputStream out = fs.create(new Path(path.pathname), 
                                             new FsPermission(mode),
                                             overwrite,
                                             bufferSize,
                                             replication,
                                             blockSize,
                                             null); // progress
          long id = insert(out);
          ThriftHandle obj = new ThriftHandle(id);
          HadoopThriftHandler.LOG.debug("created: " + path + " id: " + id);
          return obj;
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
       * Opens an existing file and returns a handle to read it
       */
      public ThriftHandle open(Pathname path) throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("open: " + path);
          FSDataInputStream out = fs.open(new Path(path.pathname));
          long id = insert(out);
          ThriftHandle obj = new ThriftHandle(id);
          HadoopThriftHandler.LOG.debug("opened: " + path + " id: " + id);
          return obj;
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
       * Opens an existing file to append to it.
       */
      public ThriftHandle append(Pathname path) throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("append: " + path);
          FSDataOutputStream out = fs.append(new Path(path.pathname));
          long id = insert(out);
          ThriftHandle obj = new ThriftHandle(id);
          HadoopThriftHandler.LOG.debug("appended: " + path + " id: " + id);
          return obj;
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
       * write to a file
       */
      public boolean write(ThriftHandle tout, String data) throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("write: " + tout.id);
          FSDataOutputStream out = (FSDataOutputStream)lookup(tout.id);
          byte[] tmp = data.getBytes("UTF-8");
          out.write(tmp, 0, tmp.length);
          HadoopThriftHandler.LOG.debug("wrote: " + tout.id);
          return true;
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
       * read from a file
       */
      public String read(ThriftHandle tout, long offset,
                         int length) throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("read: " + tout.id +
                                       " offset: " + offset +
                                       " length: " + length);
          FSDataInputStream in = (FSDataInputStream)lookup(tout.id);
          
          if (in.getPos() != offset) {
            in.seek(offset);
          }
          byte[] tmp = new byte[length];
          int numbytes = in.read(offset, tmp, 0, length);
          HadoopThriftHandler.LOG.debug("read done: " + tout.id);
          return new String(tmp, 0, numbytes, "UTF-8");
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
       * Delete a file/directory
       */
      public boolean rm(Pathname path, boolean recursive) 
                            throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("rm: " + path +
                                       " recursive: " + recursive);
          boolean ret = fs.delete(new Path(path.pathname), recursive);
          HadoopThriftHandler.LOG.debug("rm: " + path);
          return ret;
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
       * Move a file/directory
       */
      public boolean rename(Pathname path, Pathname dest) 
                            throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("rename: " + path +
                                       " destination: " + dest);
          boolean ret = fs.rename(new Path(path.pathname), 
                                  new Path(dest.pathname));
          HadoopThriftHandler.LOG.debug("rename: " + path);
          return ret;
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
       *  close file
       */
       public boolean close(ThriftHandle tout) throws ThriftIOException {
         try {
           now.set(now());
           HadoopThriftHandler.LOG.debug("close: " + tout.id);
           Object obj = remove(tout.id);
           
           if (obj instanceof FSDataOutputStream) {
             FSDataOutputStream out = (FSDataOutputStream)obj;
             out.close();
           } else if (obj instanceof FSDataInputStream) {
             FSDataInputStream in = (FSDataInputStream)obj;
             in.close();
           } else {
             throw new ThriftIOException("Unknown thrift handle.");
           }
           HadoopThriftHandler.LOG.debug("closed: " + tout.id);
           return true;
         } catch (IOException e) {
           throw new ThriftIOException(e.getMessage());
         }
       }

       /**
        * Create a directory
        */
      public boolean mkdirs(Pathname path) throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("mkdirs: " + path);
          boolean ret = fs.mkdirs(new Path(path.pathname));
          HadoopThriftHandler.LOG.debug("mkdirs: " + path);
          return ret;
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
       * Does this pathname exist?
       */
      public boolean exists(Pathname path) throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("exists: " + path);
          boolean ret = fs.exists(new Path(path.pathname));
          HadoopThriftHandler.LOG.debug("exists done: " + path);
          return ret;
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
       * Returns status about the specified pathname
       */
      public org.apache.hadoop.thriftfs.api.FileStatus stat(
                              Pathname path) throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("stat: " + path);
          org.apache.hadoop.fs.FileStatus stat = fs.getFileStatus(
                                             new Path(path.pathname));
          HadoopThriftHandler.LOG.debug("stat done: " + path);
          return new org.apache.hadoop.thriftfs.api.FileStatus(
            stat.getPath().toString(),
            stat.getLen(),
            stat.isDir(),
            stat.getReplication(),
            stat.getBlockSize(),
            stat.getModificationTime(),
            stat.getPermission().toString(),
            stat.getOwner(),
            stat.getGroup());
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
       * If the specified pathname is a directory, then return the
       * list of pathnames in this directory
       */
      public List<org.apache.hadoop.thriftfs.api.FileStatus> listStatus(
                              Pathname path) throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("listStatus: " + path);

          org.apache.hadoop.fs.FileStatus[] stat = fs.listStatus(
                                             new Path(path.pathname));
          HadoopThriftHandler.LOG.debug("listStatus done: " + path);
          org.apache.hadoop.thriftfs.api.FileStatus tmp;
          List<org.apache.hadoop.thriftfs.api.FileStatus> value = 
            new LinkedList<org.apache.hadoop.thriftfs.api.FileStatus>();

          for (int i = 0; i < stat.length; i++) {
            tmp = new org.apache.hadoop.thriftfs.api.FileStatus(
                        stat[i].getPath().toString(),
                        stat[i].getLen(),
                        stat[i].isDir(),
                        stat[i].getReplication(),
                        stat[i].getBlockSize(),
                        stat[i].getModificationTime(),
                        stat[i].getPermission().toString(),
                        stat[i].getOwner(),
                        stat[i].getGroup());
            value.add(tmp);
          }
          return value;
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
       * Sets the permission of a pathname
       */
      public void chmod(Pathname path, short mode) throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("chmod: " + path + 
                                       " mode " + mode);
          fs.setPermission(new Path(path.pathname), new FsPermission(mode));
          HadoopThriftHandler.LOG.debug("chmod done: " + path);
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
       * Sets the owner & group of a pathname
       */
      public void chown(Pathname path, String owner, String group) 
                                                         throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("chown: " + path +
                                       " owner: " + owner +
                                       " group: " + group);
          fs.setOwner(new Path(path.pathname), owner, group);
          HadoopThriftHandler.LOG.debug("chown done: " + path);
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }

      /**
       * Sets the replication factor of a file
       */
      public void setReplication(Pathname path, short repl) throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("setrepl: " + path +
                                       " replication factor: " + repl);
          fs.setReplication(new Path(path.pathname), repl);
          HadoopThriftHandler.LOG.debug("setrepl done: " + path);
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }

      }

      /**
       * Returns the block locations of this file
       */
      public List<org.apache.hadoop.thriftfs.api.BlockLocation> 
               getFileBlockLocations(Pathname path, long start, long length) 
                                           throws ThriftIOException {
        try {
          now.set(now());
          HadoopThriftHandler.LOG.debug("getFileBlockLocations: " + path);

          org.apache.hadoop.fs.FileStatus status = fs.getFileStatus(
                                                   new Path(path.pathname));

          org.apache.hadoop.fs.BlockLocation[] stat = 
              fs.getFileBlockLocations(status, start, length);
          HadoopThriftHandler.LOG.debug("getFileBlockLocations done: " + path);

          org.apache.hadoop.thriftfs.api.BlockLocation tmp;
          List<org.apache.hadoop.thriftfs.api.BlockLocation> value = 
            new LinkedList<org.apache.hadoop.thriftfs.api.BlockLocation>();

          for (int i = 0; i < stat.length; i++) {

            // construct the list of hostnames from the array returned
            // by HDFS
            List<String> hosts = new LinkedList<String>();
            String[] hostsHdfs = stat[i].getHosts();
            for (int j = 0; j < hostsHdfs.length; j++) {
              hosts.add(hostsHdfs[j]);
            }

            // construct the list of host:port from the array returned
            // by HDFS
            List<String> names = new LinkedList<String>();
            String[] namesHdfs = stat[i].getNames();
            for (int j = 0; j < namesHdfs.length; j++) {
              names.add(namesHdfs[j]);
            }
            tmp = new org.apache.hadoop.thriftfs.api.BlockLocation(
                        hosts, names, stat[i].getOffset(), stat[i].getLength());
            value.add(tmp);
          }
          return value;
        } catch (IOException e) {
          throw new ThriftIOException(e.getMessage());
        }
      }
    }

    // Bind to port. If the specified port is 0, then bind to random port.
    private ServerSocket createServerSocket(int port) throws IOException {
      try {
        ServerSocket sock = new ServerSocket();
        // Prevent 2MSL delay problem on server restarts
        sock.setReuseAddress(true);
        // Bind to listening port
        if (port == 0) {
          sock.bind(null);
          serverPort = sock.getLocalPort();
        } else {
          sock.bind(new InetSocketAddress(port));
        }
        return sock;
      } catch (IOException ioe) {
        throw new IOException("Could not create ServerSocket on port " + port + "." +
                              ioe);
      }
    }

    /**
     * Constrcts a server object
     */
    public HadoopThriftServer(String [] args) {

      if (args.length > 0) {
        serverPort = new Integer(args[0]);
      }
      try {
        ServerSocket ssock = createServerSocket(serverPort);
        TServerTransport serverTransport = new TServerSocket(ssock);
        Iface handler = new HadoopThriftHandler("hdfs-thrift-dhruba");
        ThriftHadoopFileSystem.Processor processor = new ThriftHadoopFileSystem.Processor(handler);
        TThreadPoolServer.Options options = new TThreadPoolServer.Options();
        options.minWorkerThreads = 10;
        server = new TThreadPoolServer(processor, serverTransport,
                                               new TTransportFactory(),
                                               new TTransportFactory(),
                                               new TBinaryProtocol.Factory(),
                                               new TBinaryProtocol.Factory(), 
                                               options);
        System.out.println("Starting the hadoop thrift server on port [" + serverPort + "]...");
        HadoopThriftHandler.LOG.info("Starting the hadoop thrift server on port [" +serverPort + "]...");
        System.out.flush();

      } catch (Exception x) {
        x.printStackTrace();
      }
    }

    public static void main(String [] args) {
      HadoopThriftServer me = new HadoopThriftServer(args);
      me.server.serve();
    }
}

安定した。気がする。。
いつか問題となるようなコードを書いてpatchを作ろう