diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,5 @@
+# Revision history for hadoop-streaming
+
+## 0.1.0.0 -- 2020-04-01
+
+* First version. Released on an unsuspecting world.
diff --git a/LICENSE b/LICENSE
new file mode 100644
--- /dev/null
+++ b/LICENSE
@@ -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.
diff --git a/README.md b/README.md
new file mode 100644
--- /dev/null
+++ b/README.md
@@ -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.
diff --git a/Setup.hs b/Setup.hs
new file mode 100644
--- /dev/null
+++ b/Setup.hs
@@ -0,0 +1,2 @@
+import Distribution.Simple
+main = defaultMain
diff --git a/hadoop-streaming.cabal b/hadoop-streaming.cabal
new file mode 100644
--- /dev/null
+++ b/hadoop-streaming.cabal
@@ -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.*
diff --git a/src/HadoopStreaming.hs b/src/HadoopStreaming.hs
new file mode 100644
--- /dev/null
+++ b/src/HadoopStreaming.hs
@@ -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)]
diff --git a/test/hspec/HadoopStreamingSpec.hs b/test/hspec/HadoopStreamingSpec.hs
new file mode 100644
--- /dev/null
+++ b/test/hspec/HadoopStreamingSpec.hs
@@ -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
diff --git a/test/hspec/Main.hs b/test/hspec/Main.hs
new file mode 100644
--- /dev/null
+++ b/test/hspec/Main.hs
@@ -0,0 +1,1 @@
+{-# OPTIONS_GHC -F -pgmF hspec-discover #-}