hadoop-tools 0.3 → 0.4
raw patch · 2 files changed
+84/−59 lines, 2 filesdep +regex-pcre-builtindep +stmdep ~hadoop-rpc
Dependencies added: regex-pcre-builtin, stm
Dependency ranges changed: hadoop-rpc
Files
- hadoop-tools.cabal +3/−1
- src/Main.hs +81/−58
hadoop-tools.cabal view
@@ -1,5 +1,5 @@ name: hadoop-tools-version: 0.3+version: 0.4 synopsis: Fast command line tools for working with Hadoop. description: Fast command line tools for working with Hadoop. homepage: http://github.com/jystic/hadoop-tools@@ -35,7 +35,9 @@ , old-locale >= 1.0 , optparse-applicative >= 0.10 , protobuf >= 0.2.0.4+ , regex-pcre-builtin >= 0.94 , split >= 0.2+ , stm >= 2.4 , text >= 1.1 , time >= 1.4 , transformers >= 0.4
src/Main.hs view
@@ -3,7 +3,8 @@ module Main (main) where -import Control.Exception (SomeException, bracket)+import Control.Concurrent.STM+import Control.Exception (SomeException, throwIO, bracket, fromException) import Control.Monad import Control.Monad.Catch (handle, throwM) import Control.Monad.IO.Class (MonadIO, liftIO)@@ -20,25 +21,27 @@ import Data.Time import Data.Time.Clock.POSIX import qualified Data.Vector as V-import Data.Word (Word16, Word32, Word64)+import Data.Version (showVersion)+import Data.Word (Word16, Word64) import qualified Data.Configurator as C import Data.Configurator.Types (Worth(..))+import Options.Applicative hiding (Success) import System.Environment (getEnv)-import System.FilePath.Posix+import qualified System.FilePath as FilePath+import qualified System.FilePath.Posix as Posix import System.IO import System.IO.Unsafe (unsafePerformIO) import System.Locale (defaultTimeLocale) import Text.PrettyPrint.Boxes hiding ((<>), (//)) import Data.Hadoop.Configuration (getHadoopConfig)+import Data.Hadoop.HdfsPath import Data.Hadoop.Types import Network.Hadoop.Hdfs hiding (runHdfs) import Network.Hadoop.Read -import Options.Applicative hiding (Success)--import Data.Version (showVersion)+import qualified Glob as Glob import Paths_hadoop_tools (version) ------------------------------------------------------------------------@@ -77,7 +80,7 @@ configPath :: FilePath configPath = unsafePerformIO $ do home <- getEnv "HOME"- return (home </> ".hh")+ return (home `FilePath.combine` ".hh") {-# NOINLINE configPath #-} getHdfsUser :: IO (Maybe User)@@ -103,11 +106,11 @@ workingDirConfigPath :: FilePath workingDirConfigPath = unsafePerformIO $ do home <- getEnv "HOME"- return (home </> ".hhwd")+ return (home `FilePath.combine` ".hhwd") {-# NOINLINE workingDirConfigPath #-} getDefaultWorkingDir :: MonadIO m => m HdfsPath-getDefaultWorkingDir = liftIO $ (("/user" //) . T.encodeUtf8 . hcUser) <$> getConfig+getDefaultWorkingDir = liftIO $ (("/user" </>) . T.encodeUtf8 . hcUser) <$> getConfig getWorkingDir :: MonadIO m => m HdfsPath getWorkingDir = liftIO $ handle onError@@ -126,7 +129,7 @@ where getPath = if "/" `B.isPrefixOf` path then return path- else getWorkingDir >>= \pwd -> return (pwd // path)+ else getWorkingDir >>= \pwd -> return (pwd </> path) normalizePath :: HdfsPath -> HdfsPath normalizePath = B.intercalate "/" . dropAbsParentDir . B.split '/'@@ -137,6 +140,7 @@ where go [] (".." : ys) = go [] ys go (_ : xs) (".." : ys) = go xs ys+ go xs ("." : ys) = go xs ys go xs (y : ys) = go (y : xs) ys go xs [] = (xs, []) @@ -163,7 +167,7 @@ , subChDir , subChMod , subDiskUsage- -- , subFind+ , subFind , subGet , subList , subMkDir@@ -218,8 +222,8 @@ modifyPerms :: Word16 -> HdfsPath -> Hdfs () modifyPerms mode path = do absPath <- getAbsolute path- info <- getFileInfo absPath- case info of+ minfo <- getFileInfo absPath+ case minfo of Nothing -> fail $ unwords ["No such file", B.unpack absPath] Just FileStatus{..} -> do {-@@ -238,14 +242,18 @@ subFind :: SubCommand subFind = SubCommand "find" "Recursively search a directory tree" go where- go = find <$> (argument bstr (completeDir <> help "the path to recursively search"))+ go = find <$> (optional (argument bstr (completeDir <> help "the path to recursively search"))) <*> (optional (option bstr (long "name" <> metavar "FILENAME"- <> help "the file name to match exactly")))- find path mexpr = SubHdfs $ do- absPath <- getAbsolute path- printFindResults absPath $ maybe (const True) feq mexpr+ <> help "the file name to match")))+ find mpath mexpr = SubHdfs $ do+ matcher <- liftIO (mkMatcher mexpr)+ printFindResults (fromMaybe "" mpath) matcher - feq expr FileStatus{..} = expr == fsPath+ mkMatcher :: Maybe ByteString -> IO (FileStatus -> Bool)+ mkMatcher Nothing = return (const True)+ mkMatcher (Just expr) = do+ glob <- Glob.compile expr+ return (Glob.matches glob . fsPath) subGet :: SubCommand subGet = SubCommand "get" "Get a file" go@@ -253,7 +261,7 @@ go = get <$> argument bstr (completePath <> help "source file") <*> optional (argument str (completePath <> help "destination file")) get src mdst = SubHdfs $ do- let dst = fromMaybe (takeFileName $ B.unpack src) mdst+ let dst = fromMaybe (Posix.takeFileName $ B.unpack src) mdst absSrc <- getAbsolute src mReadHandle <- openRead absSrc let doRead readHandle = liftIO $ bracket@@ -265,7 +273,7 @@ subList :: SubCommand subList = SubCommand "ls" "List the contents of a directory" go where- go = ls <$> optional (argument bstr (completeDir <> help "the directory to list"))+ go = ls <$> optional (argument bstr (completePath <> help "the directory to list")) ls path = SubHdfs $ printListing =<< getAbsolute (fromMaybe "" path) subMkDir :: SubCommand@@ -312,9 +320,9 @@ ------------------------------------------------------------------------ fileCompletion :: (FileType -> Bool) -> Completer-fileCompletion p = mkCompleter $ \spath -> handle ignore $ runHdfs $ do- let dir = B.pack $ fst $ splitFileName' spath- path = B.pack spath+fileCompletion p = mkCompleter $ \strPath -> handle ignore $ runHdfs $ do+ let path = B.pack strPath+ dir = takeParent path ls <- getListing' =<< getAbsolute dir @@ -327,12 +335,14 @@ where ignore (RemoteError _ _) = return [] - splitFileName' x = case splitFileName x of- ("./", f) -> ("", f)- (d, f) -> (d, f)+takeParent :: HdfsPath -> HdfsPath+takeParent bs = case B.elemIndexEnd '/' bs of+ Nothing -> B.empty+ Just 0 -> "/"+ Just ix -> B.take ix bs displayPath :: HdfsPath -> FileStatus -> HdfsPath-displayPath parent file = parent // fsPath file <> suffix+displayPath parent file = parent </> fsPath file <> suffix where suffix = case fsFileType file of Dir -> "/"@@ -373,21 +383,50 @@ <+> col left (T.unpack . fsGroup) <+> col right (formatSize . fsLength) <+> col right (formatUTC . getModTime)- <+> col left (T.unpack . T.decodeUtf8 . fsPath)+ <+> col left (ifEmpty basePath . T.unpack . T.decodeUtf8 . fsPath)+ where+ ifEmpty def x = if x=="" then def else x+ basePath = Posix.takeFileName (B.unpack path) printFindResults :: HdfsPath -> (FileStatus -> Bool) -> Hdfs ()-printFindResults path cond = handle (liftIO . printError) $ do- ls <- getListingOrFail path- V.mapM_ printMatch ls+printFindResults path cond = do+ absPath <- getAbsolute path+ q <- getListingRecursive absPath+ liftIO $ loop q $ printMatch (replaceFullPathWithInputPath absPath path) where- printMatch :: FileStatus -> Hdfs ()- printMatch fs@FileStatus{..} = do- let path' = displayPath path fs- when (cond fs) (liftIO $ B.putStrLn path')- case fsFileType of- Dir -> printFindResults path' cond- _ -> return ()+ loop :: TBQueue (Maybe (HdfsPath, Either SomeException (V.Vector FileStatus)))+ -> (HdfsPath -> FileStatus -> IO ())+ -> IO ()+ loop q io = do+ mx <- atomically (readTBQueue q)+ case mx of+ Nothing -> return ()+ Just (parent, x) -> do+ case x of+ (Left err) -> printOrThrow err+ (Right ls) -> V.mapM_ (io parent) ls+ loop q io + printOrThrow :: SomeException -> IO ()+ printOrThrow ex = case fromException ex of+ Nothing -> throwIO ex+ Just err -> printError err++ printMatch :: (HdfsPath -> HdfsPath) -> HdfsPath -> FileStatus -> IO ()+ printMatch fixup parent fs@FileStatus{..} = do+ let path' = fixup (displayPath parent fs)+ when (cond fs) (B.putStrLn path')++replaceFullPathWithInputPath :: HdfsPath -> HdfsPath -> HdfsPath -> HdfsPath+replaceFullPathWithInputPath fullPath inputPath path+ | fullPath' `B.isPrefixOf` path = inputPath </> B.drop (B.length fullPath') path+ | otherwise = path+ where+ fullPath' = addDirSlash fullPath++ addDirSlash dir | "/" `B.isSuffixOf` dir = dir+ | otherwise = dir <> "/"+ ------------------------------------------------------------------------ formatSize :: Word64 -> String@@ -426,9 +465,9 @@ printError :: RemoteError -> IO () printError (RemoteError subject body)- | oneLiner = T.putStrLn firstLine- | T.null body = T.putStrLn subject- | otherwise = T.putStrLn subject >> T.putStrLn body+ | oneLiner = T.hPutStrLn stderr firstLine+ | T.null body = T.hPutStrLn stderr subject+ | otherwise = T.hPutStrLn stderr subject >> T.putStrLn body where oneLiner = subject `elem` [ "org.apache.hadoop.security.AccessControlException" , "org.apache.hadoop.fs.FileAlreadyExistsException"@@ -437,22 +476,6 @@ isAccessDenied :: RemoteError -> Bool isAccessDenied (RemoteError s _) = s == "org.apache.hadoop.security.AccessControlException"----------------------------------------------------------------------------infixr 5 //--(//) :: HdfsPath -> HdfsPath -> HdfsPath-(//) xs ys | B.null xs = ys- | B.null ys = xs- | B.head ys == '/' = ys- | B.last xs == '/' = xs <> ys- | otherwise = xs <> "/" <> ys--trimEnd :: Char -> ByteString -> ByteString-trimEnd b bs | B.null bs = B.empty- | B.last bs == b = trimEnd b (B.init bs)- | otherwise = bs ------------------------------------------------------------------------