hadoop-streaming (empty) → 0.1.0.0

raw patch · 8 files changed

+353/−0 lines, 8 filesdep +basedep +conduitdep +extrasetup-changed

Dependencies added: base, conduit, extra, hadoop-streaming, hspec, text

Files

• CHANGELOG.md +5/−0
• LICENSE +29/−0
• README.md +3/−0
• Setup.hs +2/−0
• hadoop-streaming.cabal +61/−0
• src/HadoopStreaming.hs +167/−0
• test/hspec/HadoopStreamingSpec.hs +85/−0
• test/hspec/Main.hs +1/−0

@@ -0,0 +1,5 @@+# Revision history for hadoop-streaming++## 0.1.0.0 -- 2020-04-01++* First version. Released on an unsuspecting world.

@@ -0,0 +1,29 @@+BSD 3-Clause License++Copyright (c) 2020, Ziyang Liu+All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++1. Redistributions of source code must retain the above copyright notice, this+   list of conditions and the following disclaimer.++2. Redistributions in binary form must reproduce the above copyright notice,+   this list of conditions and the following disclaimer in the documentation+   and/or other materials provided with the distribution.++3. Neither the name of the copyright holder nor the names of its+   contributors may be used to endorse or promote products derived from+   this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

@@ -0,0 +1,3 @@+A simple Hadoop streaming library based on [conduit](https://hackage.haskell.org/package/conduit),+useful for writing mapper and reducer logic in Haskell and running it on AWS Elastic MapReduce,+Azure HDInsight, GCP Dataproc, and so forth.

@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain

@@ -0,0 +1,61 @@+cabal-version:       2.4++name:                hadoop-streaming+version:             0.1.0.0+synopsis:            A simple Hadoop streaming library+description:+  A simple Hadoop streaming library based on <https://hackage.haskell.org/package/conduit conduit>,+  useful for writing mapper and reducer logic in Haskell and running it on AWS Elastic MapReduce,+  Azure HDInsight, GCP Dataproc, and so forth.+category:            Cloud, Distributed Computing, MapReduce+homepage:            https://github.com/zliu41/hadoop-streaming+bug-reports:         https://github.com/zliu41/hadoop-streaming/issues+license:             BSD-3-Clause+license-file:        LICENSE+author:              Ziyang Liu <free@cofree.io>+maintainer:          Ziyang Liu <free@cofree.io>+copyright:           2020 Ziyang Liu+build-type:          Simple+tested-with:         GHC==8.8.3, GHC==8.6.5++extra-source-files:+    CHANGELOG.md+    README.md++source-repository head+  type: git+  location: https://github.com/zliu41/hadoop-streaming++library+  exposed-modules:+      HadoopStreaming+  other-modules:+      Paths_hadoop_streaming+  autogen-modules:+      Paths_hadoop_streaming+  hs-source-dirs:+      src+  build-depends:+      base >=4.12 && <5,+      conduit >= 1.3.1 && <1.4,+      extra >= 1.6.18 && <1.8,+      text >=1.2.4.0 && <1.3+  default-language: Haskell2010++test-suite hspec+  type: exitcode-stdio-1.0+  main-is: Main.hs+  other-modules:+      HadoopStreamingSpec+  hs-source-dirs:+      test/hspec+  ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N+  build-depends:+      base,+      conduit,+      extra,+      hspec,+      text,+      hadoop-streaming+  default-language: Haskell2010+  build-tool-depends: hspec-discover:hspec-discover == 2.*

@@ -0,0 +1,167 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE ScopedTypeVariables #-}++module HadoopStreaming+  ( Mapper(..)+  , Reducer(..)+  , runMapper+  , runMapperWith+  , runReducer+  , runReducerWith+  , println+  , incCounter+  , incCounterBy+  , sourceHandle+  , sinkHandle+  ) where++import           Control.Monad.Extra (unlessM)+import           Control.Monad.IO.Class (MonadIO(..))+import           Data.Conduit (ConduitT, runConduit, (.|))+import qualified Data.Conduit as C+import qualified Data.Conduit.Combinators as C+import qualified Data.Conduit.List as CL+import           Data.Text (Text)+import qualified Data.Text as Text+import qualified Data.Text.IO as Text+import qualified System.IO as IO+++-- | A @Mapper@ consists of a decoder, an encoder, and a stream that transforms+-- each input into a @(key, value)@ pair.+data Mapper e m = forall input k v. Mapper+  (Text -> Either e input)+  -- ^ Decoder for mapper input.+  (k -> v -> Text)+  -- ^ Encoder for mapper output.+  (ConduitT input (k, v) m ())+  -- ^ A stream transforming @input@ into @(k, v)@ pairs.++-- | A @Reducer@ consists of a decoder, an encoder, and a stream that transforms+-- each key and all values associated with the key into zero or more @res@.+data Reducer e m = forall k v res. Eq k => Reducer+  (Text -> Either e (k, v))+  -- ^ Decoder for reducer input+  (res -> Text)+  -- ^ Encoder for reducer output+  (k -> v -> ConduitT v res m ())+  -- ^ A stream processing a key and all values associated with the key. The parameter+  -- @v@ is the first value associated with the key (since a key always has one or more+  -- values), and the remaining values are processed by the conduit.+  --+  -- Examples:+  --+  -- @+  -- import qualified Data.Conduit as C+  -- import qualified Data.Conduit.Combinators as C+  --+  -- -- Sum up all values associated with the key and emit a (key, sum) pair.+  -- sumValues :: (Monad m, Num v) => k -> v -> ConduitT v (k, v) m ()+  -- sumValues k v0 = C.foldl (+) v0 >>= C.yield . (k,)+  --+  -- -- Increment a counter for each (key, value) pair, and emit the (key, value) pair.+  -- incCounterAndEmit :: MonadIO m => k -> v -> ConduitT v (k, v) m ()+  -- incCounterAndEmit k v0 = C.leftover v0 <> C.mapM \\v ->+  --   incCounter "reducer" "key-value pairs" >> pure (k, v)+  -- @++-- | Run a 'Mapper'. Takes input from 'stdin' and emits the result to 'stdout'.+--+-- > runMapper = runMapperWith (sourceHandle stdin) (sinkHandle stdout)+runMapper+  :: MonadIO m+  => (Text -> e -> m ())+  -- ^ A action to be executed for each input that cannot be decoded. The first parameter+  -- is the input and the second parameter is the decoding error. One may choose to, for instance,+  -- increment a counter and 'println' an error message.+  -> Mapper e m -> m ()+runMapper = runMapperWith stdin stdout++-- | Like 'runMapper', but allows specifying a source and a sink.+--+-- > runMapper = runMapperWith (sourceHandle stdin) (sinkHandle stdout)+runMapperWith+  :: MonadIO m+  => ConduitT () Text m ()+  -> ConduitT Text C.Void m ()+  -> (Text -> e -> m ())+  -> Mapper e m -> m ()+runMapperWith source sink f (Mapper dec enc trans) = runConduit $+    source .| CL.mapMaybeM g .| trans .| C.map (uncurry enc) .| sink+  where+    g input = either ((Nothing <$) . f input) (pure . Just) (dec input)++-- | Run a 'Reducer'. Takes input from 'stdin' and emits the result to 'stdout'.+--+-- > runReducer = runReducerWith (sourceHandle stdin) (sinkHandle stdout)+runReducer+  :: MonadIO m+  => (Text -> e -> m ())+  -- ^ A action to be executed for each input that cannot be decoded. The first parameter+  -- is the input and the second parameter is the decoding error. One may choose to, for instance,+  -- increment a counter and 'println' an error message.+  -> Reducer e m -> m ()+runReducer = runReducerWith stdin stdout++-- | Like 'runReducer', but allows specifying a source and a sink.+--+-- > runReducer = runReducerWith (sourceHandle stdin) (sinkHandle stdout)+runReducerWith+  :: MonadIO m+  => ConduitT () Text m ()+  -> ConduitT Text C.Void m ()+  -> (Text -> e -> m ())+  -> Reducer e m -> m ()+runReducerWith source sink f (Reducer dec enc trans) = runConduit $+    source .| CL.mapMaybeM g .| CL.groupOn1 fst .| reduceKey trans .| C.map enc .| sink+  where+    g input = either ((Nothing <$) . f input) (pure . Just) (dec input)++reduceKey :: forall m k v res. Monad m+          => (k -> v -> ConduitT v res m ())+          -> ConduitT ((k,v), [(k,v)]) res m ()+reduceKey f = go+  where+    go = C.await >>= maybe (pure ()) \((k, v), kvs) ->+      C.yieldMany (map snd kvs) .| f k v >> go++stdin :: MonadIO m => ConduitT i Text m ()+stdin = sourceHandle IO.stdin++stdout :: MonadIO m => ConduitT Text o m ()+stdout = sinkHandle IO.stdout++-- | Stream the contents of a 'IO.Handle' one line at a time as 'Text'.+sourceHandle :: MonadIO m => IO.Handle -> ConduitT i Text m ()+sourceHandle h = go .| C.filter (not . Text.all (== ' '))+  where+    go = unlessM (liftIO (IO.hIsEOF h)) (liftIO (Text.hGetLine h) >>= C.yield >> go)++-- | Stream data to a 'IO.Handle', separated by @\\n@.+sinkHandle :: MonadIO m => IO.Handle -> ConduitT Text o m ()+sinkHandle h = C.awaitForever (liftIO . Text.hPutStrLn h)++-- | Like 'Text.putStrLn', but writes to 'stderr'.+println :: MonadIO m => Text -> m ()+println = liftIO . Text.hPutStrLn IO.stderr++-- | Increment a counter by 1.+incCounter+  :: MonadIO m+  => Text -- ^ Group name. Must not contain comma.+  -> Text -- ^ Counter name. Must not contain comma.+  -> m ()+incCounter = incCounterBy 1++-- | Increment a counter by @n@.+incCounterBy+  :: MonadIO m+  => Int+  -> Text -- ^ Group name. Must not contain comma.+  -> Text -- ^ Counter name. Must not contain comma.+  -> m ()+incCounterBy n group name = println $+  "reporter:counter:" <> Text.intercalate "," [group, name, Text.pack (show n)]

@@ -0,0 +1,85 @@+{-# LANGUAGE TupleSections, ViewPatterns #-}++module HadoopStreamingSpec where++import qualified Data.Conduit as C+import qualified Data.Conduit.Combinators as C+import Data.Text (Text)+import qualified Data.Text as Text+import System.IO.Extra++import HadoopStreaming++import Test.Hspec+++ignoreDecodeError :: a -> b -> IO ()+ignoreDecodeError _ _ = pure ()++spec :: Spec+spec =+  describe "Testing HadoopStreaming" $ do+    it "test case 1 - mapper" $ do+      let fin = "test/resource/1.in"+          fout = "test/resource/1.mapper-out"+      actual <- withFile fin ReadMode $ \hin ->+        withTempFile $ \temp -> do+          withFile temp WriteMode $ \hout -> do+            let mapper = Mapper dec enc trans+                  where+                    dec :: Text -> Either String (Int, Int)+                    dec = Right . read . Text.unpack++                    enc :: Int -> Int -> Text+                    enc x y = Text.pack $ show x ++ ", " ++ show y++                    trans = C.map $ \(x, y) -> (x + 1000, y + 100)+            runMapperWith (sourceHandle hin) (sinkHandle hout) ignoreDecodeError mapper+          readFile temp+      expected <- readFile fout++      actual `shouldBe` expected++    it "test case 1 - reducer" $ do+      let fin = "test/resource/1.in"+          fout = "test/resource/1.reducer-out"+      actual <- withFile fin ReadMode $ \hin ->+        withTempFile $ \temp -> do+          withFile temp WriteMode $ \hout -> do+            let reducer = Reducer dec enc trans+                  where+                    dec :: Text -> Either String (Int, Int)+                    dec = Right . read . Text.unpack++                    enc :: (Int, Int) -> Text+                    enc (x, y) = Text.pack $ show x ++ ", " ++ show y++                    trans = \k v -> C.foldl (+) v >>= C.yield . (k,)+            runReducerWith (sourceHandle hin) (sinkHandle hout) ignoreDecodeError reducer+          readFile temp+      expected <- readFile fout++      actual `shouldBe` expected++    it "test case 1 - mapper - odd keys" $ do+      let fin = "test/resource/1.in"+          fout = "test/resource/1.mapper-out-oddkeys"+      actual <- withFile fin ReadMode $ \hin ->+        withTempFile $ \temp -> do+          withFile temp WriteMode $ \hout -> do+            let mapper = Mapper dec enc trans+                  where+                    dec :: Text -> Either () (Int, Int)+                    dec (read . Text.unpack -> (x, y))+                      | odd x = Right (x, y)+                      | otherwise = Left ()++                    enc :: Int -> Int -> Text+                    enc x y = Text.pack $ show x ++ ", " ++ show y++                    trans = C.map $ \(x, y) -> (x + 1000, y + 100)+            runMapperWith (sourceHandle hin) (sinkHandle hout) ignoreDecodeError mapper+          readFile temp+      expected <- readFile fout++      actual `shouldBe` expected

@@ -0,0 +1,1 @@+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}