hadoop-streaming 0.1.0.0 → 0.2.0.0

raw patch · 7 files changed

+356/−100 lines, 7 filesdep +bytestringdep ~textPVP ok

version bump matches the API change (PVP)

Dependencies added: bytestring

Dependency ranges changed: text

API changes (from Hackage documentation)

Files

• CHANGELOG.md +4/−0
• README.md +43/−0
• hadoop-streaming.cabal +5/−2
• src/HadoopStreaming.hs +56/−78
• src/HadoopStreaming/ByteString.hs +93/−0
• src/HadoopStreaming/Text.hs +134/−0
• test/hspec/HadoopStreamingSpec.hs +21/−20

@@ -1,5 +1,9 @@ # Revision history for hadoop-streaming +## 0.2.0.0 -- 2020-04-05++* Make input and output types abstract.+ ## 0.1.0.0 -- 2020-04-01  * First version. Released on an unsuspecting world.

@@ -1,3 +1,46 @@ 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.++Hackage: https://hackage.haskell.org/package/hadoop-streaming++# Word Count Example++See the [Haddock in `HadoopStreaming.Text`](https://hackage.haskell.org/package/hadoop-streaming-0.1.0.0/docs/HadoopStreaming-Text.html)+for a simple word-count example.++# A Few Things to Note++## ByteString vs Text++The `HadoopStreaming` module provides the general `Mapper` and `Reducer` data types, whose input and output+types are abstract. They are usually instantiated with either `ByteString` or `Text`.+`ByteString` is more suitable if the input/output needs to be decoded/encoded, for instance using the+`base64-bytestring` library. On the other hand, `Text` could make more sense if decoding/encoding is not needed,+or if the data is not UTF-8 encoded (see below regarding encodings). In general I'd imagine `ByteString` being+used much more often than `Text`.++The `HadoopStreaming.ByteString` and `HadoopStreaming.Text` modules provide some utilities for working with+`ByteString` and `Text`, respectively.++## Encoding++It is highly recommended that your input data be UTF-8 encoded, as this is the default encoding Hadoop uses. If you+must use other encodings such as UTF-16, keep in mind the following gotchas:++- It is not enough that your code can work with the encoding you choose to use:++  - By default, if any of your input files does not end with a UTF-8 representation of newline, i.e., a `0x0A` byte, Hadoop+    streaming will add a `0x0A` byte.++  - Likewise, if any line in your mapper output does not contain a UTF-8 representation of tab (`0x09`), Hadoop streaming will+    add it at the end of the line.++  This will almost certainly break your job. It may be possible to configure Hadoop streaming and tell it to use other encodings,+  so that the above behavior is consistent with the encoding you choose to use, but I don't know whether that is the case. I tried+  `-D mapreduce.map.java.opts="-Dfile.encoding=UTF-16BE"` but that doesn't seem to work.++- If you use `ByteString` as the input type and use `Data.ByteString.hGetLine` to read lines from the input, be+  aware that `Data.ByteString.hGetLine` uses `0x0A` bytes as line breaks, so it doesn't work properly for+  non-UTF-8 encoded input. For example, in UTF-16BE and UTF-16LE, the newline character is encoded as `0x00 0x0A` and `0x0A 0x00`,+  respectively.

@@ -1,7 +1,7 @@ cabal-version:       2.4  name:                hadoop-streaming-version:             0.1.0.0+version:             0.2.0.0 synopsis:            A simple Hadoop streaming library description:   A simple Hadoop streaming library based on <https://hackage.haskell.org/package/conduit conduit>,@@ -29,6 +29,8 @@ library   exposed-modules:       HadoopStreaming+      HadoopStreaming.ByteString+      HadoopStreaming.Text   other-modules:       Paths_hadoop_streaming   autogen-modules:@@ -37,6 +39,7 @@       src   build-depends:       base >=4.12 && <5,+      bytestring >=0.10 && <0.11,       conduit >= 1.3.1 && <1.4,       extra >= 1.6.18 && <1.8,       text >=1.2.4.0 && <1.3@@ -52,10 +55,10 @@   ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N   build-depends:       base,+      bytestring,       conduit,       extra,       hspec,-      text,       hadoop-streaming   default-language: Haskell2010   build-tool-depends: hspec-discover:hspec-discover == 2.*

@@ -2,23 +2,25 @@ {-# LANGUAGE GADTs #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE RankNTypes #-}-{-# LANGUAGE ScopedTypeVariables #-} +-----------------------------------------------------------------------------+-- |+-- Module      :  HadoopStreaming+-- Maintainer  :  Ziyang Liu <free@cofree.io>+--+-- Hadoop streaming makes it possible to write Hadoop jobs in Haskell. See the+-- <https://hadoop.apache.org/docs/current/hadoop-streaming/HadoopStreaming.html official documentation> for how+-- Hadoop streaming works. See "HadoopStreaming.Text" for a word count example. 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@@ -30,24 +32,24 @@ 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 @Mapper@ consists of a decoder, an encoder, and a stream transforming+-- input into @(key, value)@ pairs.+data Mapper i o e m = forall j k v. Mapper+  (i -> Either e j)+  -- ^ Decoder for mapper input+  (k -> v -> o)+  -- ^ Encoder for mapper output+  (ConduitT j (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))+-- | A @Reducer@ consists of a decoder, an encoder, and a stream transforming+-- each key and all values associated with the key into some result values.+data Reducer i o e m = forall k v r. Eq k => Reducer+  (i -> Either e (k, v))   -- ^ Decoder for reducer input-  (res -> Text)+  (r -> o)   -- ^ Encoder for reducer output-  (k -> v -> ConduitT v res m ())+  (k -> v -> ConduitT v r 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.@@ -68,83 +70,59 @@   --   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)++-- | Run a 'Mapper'. runMapper   :: MonadIO m-  => (Text -> e -> m ())-  -- ^ A action to be executed for each input that cannot be decoded. The first parameter+  => ConduitT () i m ()+  -- ^ Mapper source. The source should generally stream from @stdin@, and produce+  -- a value for each line of the input, as that is how Hadoop streaming is supposed+  -- to work, but this is not enforced. For example, if you really want to, you can produce a single+  -- value for the entire input split (which can be done using @Data.Conduit.Combinators.stdin@+  -- as the source). Note that regardless of what the source does, the values of Hadoop counters+  -- "Map input records" and "Reduce input records" are always the number of lines of the mapper and reducer+  -- input, because these counters are managed by the Hadoop framework.+  --+  -- An example is 'HadoopStreaming.Text.stdinLn', which streams from 'IO.stdin' as 'Text', one line+  -- at a time.+  -> ConduitT o C.Void m ()+  -- ^ Mapper sink. It should generally stream to 'IO.stdout'. An example is+  -- 'HadoopStreaming.Text.stdoutLn'.+  -> (i -> e -> m ())+  -- ^ An 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 $+  -> Mapper i o e m -> m ()+runMapper 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)+-- | Run a 'Reducer'. 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 $+  => ConduitT () i m ()+  -- ^ Reducer source+  -> ConduitT o C.Void m ()+  -- ^ Reducer sink+  -> (i -> e -> m ())+  -- ^ An action to be executed for each input that cannot be decoded.+  -> Reducer i o e m -> m ()+runReducer 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+  :: Monad m+  => (k -> v -> ConduitT v r m ())+  -> ConduitT ((k,v), [(k,v)]) r 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'.+-- | Like 'Text.putStrLn', but writes to 'IO.stderr'. println :: MonadIO m => Text -> m () println = liftIO . Text.hPutStrLn IO.stderr 

@@ -0,0 +1,93 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TupleSections #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  HadoopStreaming.ByteString+-- Maintainer  :  Ziyang Liu <free@cofree.io>+--+-- This module has some utilities for working with 'ByteString' in Hadoop streaming.+module HadoopStreaming.ByteString+  ( sourceHandle+  , sinkHandle+  , stdinLn+  , stdoutLn+  , defaultKeyValueEncoder+  , defaultKeyValueDecoder+  ) where++import           Conduit (MonadThrow(..), lift)+import           Control.Exception (IOException, try)+import           Control.Monad.Extra (unlessM, (>=>))+import           Control.Monad.IO.Class (MonadIO(..))+import           Data.Conduit (ConduitT, (.|))+import qualified Data.Conduit as C+import qualified Data.Conduit.Combinators as C+import           Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as BC+import qualified System.IO as IO+++-- | Stream the contents of a 'IO.Handle' one line at a time as 'ByteString'.+--+-- __NB__: This only works if the input from the 'IO.Handle' is UTF-8 encoded.+sourceHandle+  :: MonadIO m => IO.Handle -> ConduitT i ByteString m ()+sourceHandle h = go .| C.filter (not . BC.all (== ' '))+  where+    go = unlessM (liftIO (IO.hIsEOF h)) (liftIO (BC.hGetLine h) >>= C.yield >> go)++-- | Stream data to a 'IO.Handle', separated by @\\n@.+--+-- __NB__: This only works if the data is UTF-8 encoded.+sinkHandle :: MonadIO m => IO.Handle -> ConduitT ByteString o m ()+sinkHandle h = C.awaitForever (liftIO . BC.hPutStrLn h)++-- | Stream the contents from 'System.IO.stdin' one line at a time as 'ByteString'.+--+-- __NB__: This only works if the input from the 'IO.Handle' is UTF-8 encoded.+--+-- > stdinLn = sourceHandle System.IO.stdin+stdinLn :: (MonadIO m, MonadThrow m) => ConduitT i ByteString m ()+stdinLn = sourceHandle IO.stdin++-- | Stream data to 'System.IO.stdout', separated by @\\n@.+--+-- __NB__: This only works if the data is UTF-8 encoded.+--+-- > stdoutLn = sinkHandle System.IO.stdout+stdoutLn :: MonadIO m => ConduitT ByteString o m ()+stdoutLn = sinkHandle IO.stdout++-- | Encode a key-value pair by separating them with a (UTF-8 encoded) tab (i.e., a @0x09@ byte),+-- which is the default way the mapper output should be formatted.+defaultKeyValueEncoder+  :: (k -> ByteString)+  -- ^ Key encoder+  -> (v -> ByteString)+  -- ^ Value encoder+  -> k -> v -> ByteString+defaultKeyValueEncoder encKey encValue k v = encKey k <> "\t" <> encValue v++-- | Decode a line by treating the prefix up to the first tab as key, and the suffix after the first+-- tab as value. If the line does not contain a tab, or if the first tab is the last character,+-- the whole line is considered as key, and the value decoder is not used.+--+-- __NB__: This only works if the data is UTF-8 encoded.+defaultKeyValueDecoder+  :: (ByteString -> Either e k)+  -- ^ Key decoder+  -> (ByteString -> Either e v)+  -- ^ Value decoder+  -> ByteString -> Either e (k, Maybe v)+defaultKeyValueDecoder decKey decValue i+    | BC.length i2 <= 1 = (,Nothing) <$> decKey i1+    | otherwise = do+        k <- decKey i1+        v <- decValue (BC.tail i2)+        pure (k, Just v)+  where+    (i1, i2) = BC.break (== '\t') i

@@ -0,0 +1,134 @@+{-# LANGUAGE BlockArguments #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeApplications #-}++-----------------------------------------------------------------------------+-- |+-- Module      :  HadoopStreaming.Text+-- Maintainer  :  Ziyang Liu <free@cofree.io>+--+-- This module has some utilities for working with 'Text' in Hadoop streaming.+--+-- Word count example:+--+-- @+--  {-\# LANGUAGE OverloadedStrings, TupleSections \#-}+--+--  import Data.Conduit (ConduitT)+--  import qualified Data.Conduit as C+--  import qualified Data.Conduit.Combinators as C+--  import Data.Void (Void)+--  import HadoopStreaming+--  import qualified HadoopStreaming.Text as HT+--  import Data.Text (Text)+--  import qualified Data.Text as Text+--+--  mapper :: Mapper Text Text Void IO+--  mapper = Mapper dec enc trans+--    where+--      dec :: Text -> Either Void [Text]+--      dec = Right . Text.words+--+--      enc :: Text -> Int -> Text+--      enc = HT.defaultKeyValueEncoder id (Text.pack . show)+--+--      trans :: ConduitT [Text] (Text, Int) IO ()+--      trans = C.concatMap (map (,1))+--+--  reducer :: Reducer Text Text Void IO+--  reducer = Reducer dec enc trans+--    where+--      dec :: Text -> Either Void (Text, Int)+--      dec i = Right (i1, read . tail . Text.unpack $ i2)+--        where (i1, i2) = Text.break (== '\\t') i+--+--      enc :: (Text, Int) -> Text+--      enc (t, c) = t <> "," <> Text.pack (show c)+--+--      trans :: Text -> Int -> ConduitT Int (Text, Int) IO ()+--      trans k v0 = C.foldl (+) v0 >>= C.yield . (k,)+-- @+module HadoopStreaming.Text+  ( sourceHandle+  , sinkHandle+  , stdinLn+  , stdoutLn+  , defaultKeyValueEncoder+  , defaultKeyValueDecoder+  ) where++import           Conduit (MonadThrow(..), lift)+import           Control.Exception (IOException, try)+import           Control.Monad.Extra (unlessM, (>=>))+import           Control.Monad.IO.Class (MonadIO(..))+import           Data.Conduit (ConduitT, (.|))+import qualified Data.Conduit as C+import qualified Data.Conduit.Combinators as C+import           Data.Text (Text)+import qualified Data.Text as Text+import qualified Data.Text.IO as Text+import qualified System.IO as IO+++-- | Stream the contents of a 'IO.Handle' one line at a time as 'Text'.+sourceHandle+  :: MonadIO m+  => (IOException -> m ())+  -- ^ An action to be executed if there is an error reading the input. This is usually+  -- caused by the input having an incorrect encoding or containing corrupt data.+  -- The recommended action is to log an error message and fail the job.+  --+  -- __NB__: The stream will terminate if an error occurrs, regardless of whether this+  -- action re-throws the error or not.+  -> IO.Handle -> ConduitT i Text m ()+sourceHandle f h = go .| C.filter (not . Text.all (== ' '))+  where+    go = unlessM (liftIO (IO.hIsEOF h)) $+      liftIO (try @IOException (Text.hGetLine h)) >>= either (lift . f) (C.yield >=> const 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)++-- | Stream the contents from 'System.IO.stdin' one line at a time as 'Text'.+--+-- > stdinLn = sourceHandle throwM System.IO.stdin+stdinLn :: (MonadIO m, MonadThrow m) => ConduitT i Text m ()+stdinLn = sourceHandle throwM IO.stdin++-- | Stream data to 'System.IO.stdout', separated by @\\n@.+--+-- > stdoutLn = sinkHandle System.IO.stdout+stdoutLn :: MonadIO m => ConduitT Text o m ()+stdoutLn = sinkHandle IO.stdout++-- | Encode a key-value pair by separating them with a tab, which is the default way+-- the mapper output should be formatted.+defaultKeyValueEncoder+  :: (k -> Text)+  -- ^ Key encoder+  -> (v -> Text)+  -- ^ Value encoder+  -> k -> v -> Text+defaultKeyValueEncoder encKey encValue k v = encKey k <> "\t" <> encValue v++-- | Decode a line by treating the prefix up to the first tab as key, and the suffix after the first+-- tab as value. If the line does not contain a tab, or if the first tab is the last character,+-- the whole line is considered as key, and the value decoder is not used.+defaultKeyValueDecoder+  :: (Text -> Either e k)+  -- ^ Key decoder+  -> (Text -> Either e v)+  -- ^ Value decoder+  -> Text -> Either e (k, Maybe v)+defaultKeyValueDecoder decKey decValue i+    | Text.length i2 <= 1 = (,Nothing) <$> decKey i1+    | otherwise = do+        k <- decKey i1+        v <- decValue (Text.tail i2)+        pure (k, Just v)+  where+    (i1, i2) = Text.break (== '\t') i

@@ -4,13 +4,14 @@  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           Data.ByteString (ByteString)+import qualified Data.ByteString.Char8 as B+import           System.IO.Extra (IOMode(..), withFile, withTempFile) -import HadoopStreaming+import           HadoopStreaming+import           HadoopStreaming.ByteString -import Test.Hspec+import           Test.Hspec   ignoreDecodeError :: a -> b -> IO ()@@ -27,14 +28,14 @@           withFile temp WriteMode $ \hout -> do             let mapper = Mapper dec enc trans                   where-                    dec :: Text -> Either String (Int, Int)-                    dec = Right . read . Text.unpack+                    dec :: ByteString -> Either String (Int, Int)+                    dec = Right . read . B.unpack -                    enc :: Int -> Int -> Text-                    enc x y = Text.pack $ show x ++ ", " ++ show y+                    enc :: Int -> Int -> ByteString+                    enc x y = B.pack $ show x ++ ", " ++ show y                      trans = C.map $ \(x, y) -> (x + 1000, y + 100)-            runMapperWith (sourceHandle hin) (sinkHandle hout) ignoreDecodeError mapper+            runMapper (sourceHandle hin) (sinkHandle hout) ignoreDecodeError mapper           readFile temp       expected <- readFile fout @@ -48,14 +49,14 @@           withFile temp WriteMode $ \hout -> do             let reducer = Reducer dec enc trans                   where-                    dec :: Text -> Either String (Int, Int)-                    dec = Right . read . Text.unpack+                    dec :: ByteString -> Either String (Int, Int)+                    dec = Right . read . B.unpack -                    enc :: (Int, Int) -> Text-                    enc (x, y) = Text.pack $ show x ++ ", " ++ show y+                    enc :: (Int, Int) -> ByteString+                    enc (x, y) = B.pack $ show x ++ ", " ++ show y                      trans = \k v -> C.foldl (+) v >>= C.yield . (k,)-            runReducerWith (sourceHandle hin) (sinkHandle hout) ignoreDecodeError reducer+            runReducer (sourceHandle hin) (sinkHandle hout) ignoreDecodeError reducer           readFile temp       expected <- readFile fout @@ -69,16 +70,16 @@           withFile temp WriteMode $ \hout -> do             let mapper = Mapper dec enc trans                   where-                    dec :: Text -> Either () (Int, Int)-                    dec (read . Text.unpack -> (x, y))+                    dec :: ByteString -> Either () (Int, Int)+                    dec (read . B.unpack -> (x, y))                       | odd x = Right (x, y)                       | otherwise = Left () -                    enc :: Int -> Int -> Text-                    enc x y = Text.pack $ show x ++ ", " ++ show y+                    enc :: Int -> Int -> ByteString+                    enc x y = B.pack $ show x ++ ", " ++ show y                      trans = C.map $ \(x, y) -> (x + 1000, y + 100)-            runMapperWith (sourceHandle hin) (sinkHandle hout) ignoreDecodeError mapper+            runMapper (sourceHandle hin) (sinkHandle hout) ignoreDecodeError mapper           readFile temp       expected <- readFile fout