diff --git a/library/Potoki/Consume.hs b/library/Potoki/Consume.hs
--- a/library/Potoki/Consume.hs
+++ b/library/Potoki/Consume.hs
@@ -18,6 +18,7 @@
   printString,
   parseBytes,
   parseText,
+  concurrently,
 )
 where
 
@@ -34,6 +35,7 @@
 import qualified Data.Text.IO as K
 import qualified Control.Foldl as D
 import qualified System.Directory as G
+import qualified Potoki.Transform.Concurrency as B
 
 
 {-# INLINABLE transform #-}
@@ -177,3 +179,11 @@
 parseText :: F.Parser output -> Consume Text (Either Text output)
 parseText =
   runParseResult . F.parse
+
+{-|
+Execute a Consume concurrently and consume its results.
+-}
+{-# INLINABLE concurrently #-}
+concurrently :: Int -> Consume a b -> Consume b c -> Consume a c
+concurrently amount consume1 consume2 =
+  transform (B.concurrently amount (J.consume consume1)) consume2
diff --git a/library/Potoki/Transform.hs b/library/Potoki/Transform.hs
--- a/library/Potoki/Transform.hs
+++ b/library/Potoki/Transform.hs
@@ -11,6 +11,7 @@
   mapFilter,
   filter,
   just,
+  distinctBy,
   distinct,
   builderChunks,
   executeIO,
@@ -19,8 +20,8 @@
   parseBytes,
   parseText,
   -- * Concurrency
-  bufferize,
-  concurrently,
+  N.bufferize,
+  N.concurrently,
   -- * File IO
   deleteFile,
   appendBytesToFile,
@@ -42,6 +43,7 @@
 import qualified Data.ByteString as J
 import qualified System.Directory as I
 import qualified Control.Concurrent.Chan.Unagi.Bounded as B
+import qualified Potoki.Transform.Concurrency as N
 
 
 {-# INLINE mapFilter #-}
@@ -64,76 +66,6 @@
 takeWhile predicate =
   Transform (pure . A.takeWhile predicate)
 
-{-# INLINE bufferize #-}
-bufferize :: Int -> Transform element element
-bufferize size =
-  Transform $ \ (A.Fetch fetch) -> do
-    (inChan, outChan) <- B.newChan size
-    forkIO $ fix $ \ loop ->
-      join $ fetch
-        (B.writeChan inChan Nothing)
-        (\ !element -> B.writeChan inChan (Just element) >> loop)
-    return $ A.Fetch $ \ nil just -> fmap (maybe nil just) (B.readChan outChan)
-
-{-|
-Identity Transform, which ensures that the inputs are fetched synchronously.
-
-Useful for concurrent transforms.
--}
-{-# INLINABLE sync #-}
-sync :: Transform a a
-sync =
-  Transform $ \ (A.Fetch fetch) -> do
-    activeVar <- newMVar True
-    return $ A.Fetch $ \ nil just -> do
-      active <- takeMVar activeVar
-      if active
-        then join $ fetch
-          (do
-            putMVar activeVar False
-            return nil)
-          (\ !element -> do
-            putMVar activeVar True
-            return (just element))
-        else do
-          putMVar activeVar False
-          return nil
-
-{-|
-Execute the transform on the specified amount of threads.
-The order of the outputs produced is indiscriminate.
--}
-{-# INLINABLE concurrently #-}
-concurrently :: Int -> Transform input output -> Transform input output
-concurrently workersAmount transform =
-  sync >>>
-  concurrentlyUnsafe workersAmount transform
-
-{-# INLINE concurrentlyUnsafe #-}
-concurrentlyUnsafe :: Int -> Transform input output -> Transform input output
-concurrentlyUnsafe workersAmount (Transform syncTransformIO) =
-  Transform $ \ fetch -> do
-    outChan <- newEmptyMVar
-    replicateM_ workersAmount $ forkIO $ do
-      A.Fetch fetchIO <- syncTransformIO fetch
-      fix $ \ loop -> join $ fetchIO
-        (putMVar outChan Nothing)
-        (\ !result -> putMVar outChan (Just result) >> loop)
-    activeWorkersAmountVar <- newMVar workersAmount
-    return $ A.Fetch $ \ nil just -> fix $ \ loop -> do
-      activeWorkersAmount <- takeMVar activeWorkersAmountVar
-      if activeWorkersAmount <= 0
-        then return nil
-        else do
-          fetchResult <- takeMVar outChan
-          case fetchResult of
-            Just result -> do
-              putMVar activeWorkersAmountVar activeWorkersAmount
-              return (just result)
-            Nothing -> do
-              putMVar activeWorkersAmountVar (pred activeWorkersAmount)
-              loop
-
 {-# INLINE mapWithParseResult #-}
 mapWithParseResult :: forall input parsed. (Monoid input, Eq input) => (input -> M.IResult input parsed) -> Transform input (Either Text parsed)
 mapWithParseResult inputToResult =
@@ -226,18 +158,23 @@
   withFile path AppendMode $ \ handle -> 
   J.hPut handle bytes
 
-{-# INLINE distinct #-}
-distinct :: (Eq element, Hashable element) => Transform element element
-distinct =
+{-# INLINE distinctBy #-}
+distinctBy :: (Eq comparable, Hashable comparable) => (element -> comparable) -> Transform element element
+distinctBy f =
   Transform $ \ (A.Fetch fetch) -> do
     stateRef <- newIORef mempty
     return $ A.Fetch $ \ nil just -> fix $ \ loop -> join $ fetch (return nil) $ \ !input -> do
+      let comparable = f input
       !set <- readIORef stateRef
-      if C.member input set
+      if C.member comparable set
         then loop
         else do
-          writeIORef stateRef $! C.insert input set
+          writeIORef stateRef $! C.insert comparable set
           return (just input)
+
+{-# INLINE distinct #-}
+distinct :: (Eq element, Hashable element) => Transform element element
+distinct = distinctBy id
 
 {-# INLINE builderChunks #-}
 builderChunks :: Transform E.Builder ByteString
diff --git a/library/Potoki/Transform/Concurrency.hs b/library/Potoki/Transform/Concurrency.hs
new file mode 100644
--- /dev/null
+++ b/library/Potoki/Transform/Concurrency.hs
@@ -0,0 +1,83 @@
+module Potoki.Transform.Concurrency
+(
+  bufferize,
+  concurrently,
+)
+where
+
+import Potoki.Prelude hiding (take, takeWhile, filter)
+import Potoki.Core.Transform
+import qualified Potoki.Fetch as A
+import qualified Potoki.Core.Fetch as A
+import qualified Control.Concurrent.Chan.Unagi.Bounded as B
+
+
+{-# INLINE bufferize #-}
+bufferize :: Int -> Transform element element
+bufferize size =
+  Transform $ \ (A.Fetch fetch) -> do
+    (inChan, outChan) <- B.newChan size
+    forkIO $ fix $ \ loop ->
+      join $ fetch
+        (B.writeChan inChan Nothing)
+        (\ !element -> B.writeChan inChan (Just element) >> loop)
+    return $ A.Fetch $ \ nil just -> fmap (maybe nil just) (B.readChan outChan)
+
+{-|
+Identity Transform, which ensures that the inputs are fetched synchronously.
+
+Useful for concurrent transforms.
+-}
+{-# INLINABLE sync #-}
+sync :: Transform a a
+sync =
+  Transform $ \ (A.Fetch fetch) -> do
+    activeVar <- newMVar True
+    return $ A.Fetch $ \ nil just -> do
+      active <- takeMVar activeVar
+      if active
+        then join $ fetch
+          (do
+            putMVar activeVar False
+            return nil)
+          (\ !element -> do
+            putMVar activeVar True
+            return (just element))
+        else do
+          putMVar activeVar False
+          return nil
+
+{-|
+Execute the transform on the specified amount of threads.
+The order of the outputs produced is indiscriminate.
+-}
+{-# INLINABLE concurrently #-}
+concurrently :: Int -> Transform input output -> Transform input output
+concurrently workersAmount transform =
+  sync >>>
+  concurrentlyUnsafe workersAmount transform
+
+{-# INLINE concurrentlyUnsafe #-}
+concurrentlyUnsafe :: Int -> Transform input output -> Transform input output
+concurrentlyUnsafe workersAmount (Transform syncTransformIO) =
+  Transform $ \ fetch -> do
+    outChan <- newEmptyMVar
+    replicateM_ workersAmount $ forkIO $ do
+      A.Fetch fetchIO <- syncTransformIO fetch
+      fix $ \ loop -> join $ fetchIO
+        (putMVar outChan Nothing)
+        (\ !result -> putMVar outChan (Just result) >> loop)
+    activeWorkersAmountVar <- newMVar workersAmount
+    return $ A.Fetch $ \ nil just -> fix $ \ loop -> do
+      activeWorkersAmount <- takeMVar activeWorkersAmountVar
+      if activeWorkersAmount <= 0
+        then return nil
+        else do
+          fetchResult <- takeMVar outChan
+          case fetchResult of
+            Just result -> do
+              putMVar activeWorkersAmountVar activeWorkersAmount
+              return (just result)
+            Nothing -> do
+              putMVar activeWorkersAmountVar (pred activeWorkersAmount)
+              loop
diff --git a/potoki.cabal b/potoki.cabal
--- a/potoki.cabal
+++ b/potoki.cabal
@@ -1,7 +1,7 @@
 name:
   potoki
 version:
-  0.7
+  0.7.2
 synopsis:
   Simple streaming in IO
 description:
@@ -72,6 +72,7 @@
   other-modules:
     Potoki.Fetch
     Potoki.Prelude
+    Potoki.Transform.Concurrency
   build-depends:
     attoparsec >=0.13 && <0.15,
     base >=4.7 && <5,
diff --git a/tests/Main.hs b/tests/Main.hs
--- a/tests/Main.hs
+++ b/tests/Main.hs
@@ -74,6 +74,12 @@
       result <- C.produceAndConsume (E.list list) (D.transform A.distinct D.list)
       assertEqual "" [1,2,3,4] result
     ,
+    testCase "Distinct By" $ do
+      let
+        list = [(1, ""),(2, ""),(3, ""),(2, ""),(3, ""),(2, ""),(1, ""),(4, ""),(1, "")] :: [(Int, String)]
+      result <- C.produceAndConsume (E.list list) (D.transform (A.distinctBy fst) D.list)
+      assertEqual "" [(1, ""),(2, ""),(3, ""),(4, "")] result
+    ,
     testCase "Concurrently" $ do
       let
         list = [1..20000]