map-reduce-folds 0.1.1.1 → 0.1.2

raw patch · 4 files changed

+266/−15 lines, 4 filesdep +streamly-bytestringdep +streamly-coredep ~basedep ~discriminationdep ~hashtablesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

Dependencies added: streamly-bytestring, streamly-core

Dependency ranges changed: base, discrimination, hashtables, hedgehog, streamly, text, vector

API changes (from Hackage documentation)

Files

• ChangeLog.md +3/−0
• bench/MapReduce.hs +36/−5
• map-reduce-folds.cabal +17/−8
• src/Control/MapReduce/Engines/Streamly.hs +210/−2

@@ -1,3 +1,6 @@+* version 0.1.2+* Compatibility with >=streamly-0.9. Guarded by “streamly9” flag due to various breaking changes between streamly versions.+ * version 0.1.1.1 * Bugfix for older versions of streamly 

@@ -1,3 +1,4 @@+{-# LANGUAGE CPP #-} {-# LANGUAGE TypeApplications #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE BangPatterns #-}@@ -38,6 +39,9 @@                                                 , randomRs                                                 , randomRIO                                                 )+#if MIN_VERSION_streamly(0,9,0)+import Streamly.Data.Stream.Prelude (MonadAsync)+#endif   createPairData :: Int -> IO [(Char, Int)]@@ -134,6 +138,20 @@   ) {-# INLINE mapReduceStreamlyOrd #-} ++#if MIN_VERSION_streamly(0,9,0)+mapReduceStreamlyOrdIO :: (MonadAsync m, Foldable g) => g (Char, Int) -> m (M.Map Char Double)+mapReduceStreamlyOrdIO = FL.foldM+  (MRSL.concatStreamFoldM+    (MRSL.streamlyEngineM MRSL.groupByOrderedKeyIO+     (MR.generalizeUnpack $ MR.Filter filterPF)+      (MR.generalizeAssign $ MR.Assign id)+      (MR.generalizeReduce $ MR.ReduceFold reducePF)+    )+  )+{-# INLINE mapReduceStreamlyOrdIO #-}+#endif+ mapReduceStreamlyHash :: Foldable g => g (Char, Int) -> M.Map Char Double mapReduceStreamlyHash = FL.fold   (MRSL.concatStreamFold@@ -168,7 +186,8 @@   ) {-# INLINE mapReduceStreamlyDiscrimination #-} -+#if MIN_VERSION_streamly(0,9,0)+#else mapReduceStreamlyC   :: forall tIn tOut m g    . (MonadAsync m, Foldable g, MRSL.IsStream tIn, MRSL.IsStream tOut)@@ -183,7 +202,7 @@     )   ) {-# INLINE mapReduceStreamlyC #-}-+#endif  mapReduceVector :: Foldable g => g (Char, Int) -> M.Map Char Double mapReduceVector = FL.fold@@ -222,6 +241,11 @@   , benchPure "mapReduce (Streamly.SerialT Engine, strict map)"               (const dat)               mapReduceStreamlyOrd+#if MIN_VERSION_streamly(0,9,0)+  , benchIO "mapReduce (Streamly.SerialT Engine, strict map, streamly toMapIO)"+              (const dat)+              mapReduceStreamlyOrdIO+#endif   , benchPure "mapReduce (Streamly.SerialT Engine, strict hash map)"               (const dat)               mapReduceStreamlyHash@@ -235,7 +259,8 @@               (const dat)               mapReduceVector   ]-+#if MIN_VERSION_streamly(0,9,0)+#else benchConcurrent dat = bgroup   "Task 1, on (Char, Int). Concurrent Engines"   [ benchPure "list, parallel (6 threads)" (const dat) mapReduceListP@@ -249,6 +274,7 @@             (const dat)             (mapReduceStreamlyC @MRSL.SerialT @MRSL.AsyncT)   ]+#endif  -- a more complex row type createMapRows :: Int -> IO (Seq.Seq (M.Map T.Text Int))@@ -278,7 +304,7 @@ -- compute the average of the sum of the values in A and B for each group reduceMFold :: FL.Fold (Int, Int) Double reduceMFold = let g (x, y) = realToFrac (x + y) in FL.premap g FL.mean---reduceMFoldMap k = fmap (M.singleton k) reduceMFold  +--reduceMFoldMap k = fmap (M.singleton k) reduceMFold  -- return [(C, <A+B>)] @@ -366,4 +392,9 @@ main = do   dat  <- createPairData 100000   dat2 <- createMapRows 100000-  defaultMain [benchOne dat, benchConcurrent dat, benchTwo dat2]+  defaultMain [benchOne dat+#if MIN_VERSION_streamly(0,9,0)+#else+              , benchConcurrent dat+#endif+              , benchTwo dat2]

@@ -1,7 +1,7 @@ cabal-version:       2.2  name:                map-reduce-folds-version:             0.1.1.1+version:             0.1.2 build-type:          Simple synopsis:            foldl wrappers for map-reduce description:         map-reduce-folds simplifies the building of folds to do map-reduce style computations on collections.  It breaks the map/reduce into an unpacking step where items may be filtered, transformed or "melted" (made into several new items), an assign step where the unpacked items are assigned keys, a grouping step where the assigned items are grouped by key and then a reduce step which is applied to each grouped subset.  Tools are provided to simplify building the individual steps and then "engines" are provided for combining them into efficient folds returning an assortment of containers.  The various pieces are replicated for effectful (monadic) steps producing effectful (monadic) folds.@@ -19,17 +19,21 @@   description: Dump HTML for the core generated by GHC during compilation   default:     False +flag streamly9+    Description: require streamly >= 0.9+    Manual: True+    Default: False  source-repository head     Type: git     Location: https://github.com/adamConnerSax/map-reduce-folds  common deps-  build-depends: base                 >= 4.12.0 && < 4.17,+  build-depends: base                 >= 4.12.0 && < 5,                  containers           >= 0.5.0 && < 0.7,                  foldl                >= 1.4.5 && < 1.5,                  profunctors          >= 5.3 && < 5.7,-                 text                 >= 1.2.3 && < 1.3,+                 text                 >= 1.2.3 && < 2.1,                  unordered-containers >= 0.2.10 && < 0.3  library@@ -49,14 +53,18 @@                  , Control.MapReduce.Engines.ParallelList    build-depends:-                discrimination       >= 0.3   && < 0.5,+                discrimination       >= 0.3   && < 0.6,                 hashable             >= 1.2.4 && < 1.5,-                hashtables           >= 1.2.0.0 && < 1.3.0.0,-                vector               >= 0.12.0 && < 0.13,+                hashtables           >= 1.2.0.0 && < 1.5,+                vector               >= 0.12.0 && < 0.14,                 parallel             >= 3.2.2 && < 3.3,                 split                >= 0.2.3 && < 0.3,                 streaming            >= 0.2.2 && < 0.3,-                streamly             >= 0.7.0 && < 0.9+  if flag(streamly9)+    build-depends:+      streamly >=0.9 && <0.10, streamly-core >=0.1.0 && <0.2, streamly-bytestring >=0.2.0 && < 0.3+  else+    build-depends: streamly >=0.8 && <0.9, streamly-bytestring >=0.1.0 && <0.2    hs-source-dirs:      src   default-language:    Haskell2010@@ -70,7 +78,7 @@   ghc-options:   build-depends:                 base-              , hedgehog >= 0.6.0 && < 1.1+              , hedgehog >= 0.6.0 && < 1.3               , map-reduce-folds               , containers               , foldl@@ -91,6 +99,7 @@   build-depends:       map-reduce-folds,                        criterion ,                        deepseq ,+                       streamly >= 0.8 && < 0.10,                        random   default-language:    Haskell2010 

@@ -53,10 +53,16 @@     -- * @groupBy@ Functions   , groupByHashableKey   , groupByOrderedKey+#if MIN_VERSION_streamly(0,9,0)+  , groupByOrderedKeyIO+#endif   , groupByHashableKeyST   , groupByDiscriminatedKey      -- * Re-Exports+#if MIN_VERSION_streamly(0,9,0)+  , Stream+#else   , SerialT   , WSerialT   , AheadT@@ -65,6 +71,7 @@   , ParallelT   , MonadAsync   , IsStream+#endif   ) where @@ -85,7 +92,17 @@ import qualified Data.Maybe                    as Maybe import qualified Data.Map.Strict               as MS import qualified Data.Sequence                 as Seq-#if MIN_VERSION_streamly(0,8,0)+#if MIN_VERSION_streamly(0,9,0)+import Streamly.Data.Stream (Stream)+import qualified Streamly.Data.Stream as S+import qualified Streamly.Data.Stream.Prelude as SP+import qualified Streamly.Data.StreamK as StreamK+import qualified Streamly.Data.Fold   as SF+import qualified Streamly.Data.Unfold   as SU+import qualified Streamly.Internal.Data.Fold   as SF+import           Control.Monad (join)+import           Control.Monad.IO.Class (MonadIO)+#elif MIN_VERSION_streamly(0,8,0) import qualified Streamly.Prelude              as S import qualified Streamly.Internal.Data.Fold   as SF import           Streamly.Prelude              ( SerialT@@ -112,8 +129,18 @@                                                 ) --import qualified Streamly.Internal.Data.Parser.ParserK.Type as Streamly #endif+#if MIN_VERSION_streamly(0,9,0)+-- | convert a Control.Foldl FoldM into a Streamly.Data.Fold fold+toStreamlyFoldM :: Functor m => FL.FoldM m a b -> SF.Fold m a b+toStreamlyFoldM (FL.FoldM step start done) = SF.Fold step' (SF.Partial <$> start) done where+  step' s a = SF.Partial <$> step s a -#if MIN_VERSION_streamly(0,8,0)+-- | convert a Control.Foldl Fold into a Streamly.Data.Fold fold+toStreamlyFold :: Monad m => FL.Fold a b -> SF.Fold m a b+toStreamlyFold (FL.Fold step start done) = SF.Fold step' (pure $ SF.Partial $ start) (pure . done) where+  step' s a = pure $ SF.Partial $ step s a++#elif MIN_VERSION_streamly(0,8,0) fromEffect :: (Monad m, IsStream t) => m a -> t m a fromEffect = S.fromEffect {-# INLINE fromEffect #-}@@ -142,6 +169,186 @@ #endif  -- | unpack for streamly based map/reduce+#if MIN_VERSION_streamly(0,9,0)+unpackStreamK :: MRC.Unpack x y -> StreamK.StreamK Identity x -> S.Stream Identity y+unpackStreamK (MRC.Filter t) = S.filter t . StreamK.toStream+unpackStreamK (MRC.Unpack f) = S.concatMap (StreamK.toStream . StreamK.fromFoldable . f) . StreamK.toStream+--  S.concatMap (StreamK.toStream . StreamK.fromFoldable . f)+{-# INLINABLE unpackStreamK #-}++-- | effectful (monadic) unpack for streamly based map/reduce+unpackStreamKM :: (Monad m) => MRC.UnpackM m x y -> StreamK.StreamK m x -> S.Stream m y+unpackStreamKM (MRC.FilterM t) = S.filterM t . StreamK.toStream+unpackStreamKM (MRC.UnpackM f) = S.concatMapM (fmap (StreamK.toStream . StreamK.fromFoldable) . f) . StreamK.toStream+{-# INLINABLE unpackStreamKM #-}++-- | make a stream into an (effectful) @[]@+resultToList :: (Monad m) => S.Stream m a -> m [a]+resultToList = S.toList+{-# INLINEABLE resultToList #-}++-- | mappend all in a monoidal stream+concatStream :: (Monad m, Monoid a) => S.Stream m a -> m a+concatStream = S.fold (SF.foldl' (<>) mempty)+{-# INLINEABLE concatStream #-}++-- | mappend everything in a pure Streamly fold+concatStreamFold :: Monoid b => FL.Fold a (S.Stream Identity b) -> FL.Fold a b+concatStreamFold = fmap (runIdentity . concatStream)+{-# INLINEABLE concatStreamFold #-}++-- | mappend everything in an effectful Streamly fold.+concatStreamFoldM+  :: (Monad m, Monoid b) => FL.FoldM m a (S.Stream m b) -> FL.FoldM m a b+concatStreamFoldM = MRC.postMapM concatStream+{-# INLINEABLE concatStreamFoldM #-}++-- | mappend everything in a concurrent Streamly fold.+concatConcurrentStreamFold+  :: (Monad m, Monoid b) => FL.Fold a (S.Stream m b) -> FL.FoldM m a b+concatConcurrentStreamFold = concatStreamFoldM . FL.generalize+{-# INLINEABLE concatConcurrentStreamFold #-}++-- | map-reduce-fold builder returning a @SerialT Identity d@ result+streamlyEngine+  :: (Foldable g, Functor g)+  => (forall z . S.Stream Identity (k, z) -> S.Stream Identity (k, g z))+  -> MRE.MapReduceFold y k c (S.Stream Identity) x d+streamlyEngine groupByKey u (MRC.Assign a) r = FL.Fold+  (flip StreamK.cons)+  StreamK.nil+  ( fmap (\(k, lc) -> MRE.reduceFunction r k lc)+    . groupByKey+    . fmap a+    . unpackStreamK u+  )+{-# INLINABLE streamlyEngine #-}++-- | unpack for concurrent streamly based map/reduce+unpackConcurrentlyK+  :: (SP.MonadAsync m) => MRC.Unpack x y -> StreamK.StreamK m x -> S.Stream m y+unpackConcurrentlyK (MRC.Filter t) = S.filter t . StreamK.toStream+unpackConcurrentlyK (MRC.Unpack f) = S.concatMap ((StreamK.toStream . StreamK.fromFoldable) . f) . StreamK.toStream+{-# INLINABLE unpackConcurrentlyK #-}++-- | possibly (depending on chosen stream types) concurrent map-reduce-fold builder returning an @(Istream t, MonadAsync m) => t m d@ result+concurrentStreamlyEngine+  :: forall m g y k c x d+   . (SP.MonadAsync m, Foldable g, Functor g)+  => (forall z . S.Stream m (k, z) -> S.Stream m (k, g z))+  -> MRE.MapReduceFold y k c (S.Stream m) x d+concurrentStreamlyEngine groupByKey u (MRC.Assign a) r = FL.Fold+  (\s a' -> (pure a') `StreamK.consM` s)+  StreamK.nil+  ( S.mapM (\(k, lc) -> return $ MRE.reduceFunction r k lc)+  . groupByKey+  . S.mapM (return . a)+  . unpackConcurrentlyK u+  )+{-# INLINABLE concurrentStreamlyEngine #-}+++-- | effectful map-reduce-fold engine returning a (Istream t => t m d) result+-- The "MonadAsync" constraint here more or less requires us to run in IO, or something IO like.+streamlyEngineM+  :: (Monad m, SP.MonadAsync m, Traversable g)+  => (forall z . S.Stream m (k, z) -> S.Stream m (k, g z))+  -> MRE.MapReduceFoldM m y k c (S.Stream m) x d+streamlyEngineM groupByKey u (MRC.AssignM a) r =+  FL.generalize+    $ FL.Fold+        (flip StreamK.cons)+        StreamK.nil+        ( S.mapM (\(k, lc) -> MRE.reduceFunctionM r k lc)+        . groupByKey -- this requires a serial stream.+        . S.mapM a+        . unpackStreamKM u+        )+{-# INLINABLE streamlyEngineM #-}++toHashMap :: (Monad m, Eq k, Monoid a, Hashable k) => SF.Fold m (k, a) (HMS.HashMap k a)+toHashMap = SF.foldl' (\hm (k, a) -> HMS.insertWith (<>) k a hm) HMS.empty+{-# INLINEABLE toHashMap #-}++streamMeta :: Monad m => SF.Fold m a b -> SU.Unfold m b c -> S.Stream m a  -> S.Stream m c+streamMeta fld unfld = S.concatEffect . fmap (S.unfold unfld) . S.fold fld+{-# INLINEABLE streamMeta #-}++-- | Group streamly stream of @(k,c)@ by @hashable@ key.+groupByHashableKey+  :: (Monad m, Hashable k, Eq k)+  => S.Stream m (k, c)+  -> S.Stream m (k, Seq.Seq c)+groupByHashableKey = streamMeta toHashMap (SU.lmap HMS.toList $ SU.fromList) . fmap (second Seq.singleton)+{-# INLINABLE groupByHashableKey #-}++toMap :: (Monad m, Monoid a, Ord k) => SF.Fold m (k, a) (MS.Map k a)+toMap = SF.foldl' (\hm (k, a) -> MS.insertWith (<>) k a hm) MS.empty+{-# INLINEABLE toMap #-}++-- | Group streamly stream of @(k,c)@ by ordered key.+groupByOrderedKey+  :: (Monad m, Ord k) => S.Stream m (k, c) -> S.Stream m (k, Seq.Seq c)+groupByOrderedKey = --streamMeta (SF.toMap fst (SF.lmap snd toSeq)) (SU.lmap MS.toList $ SU.fromList)+  streamMeta toMap (SU.lmap MS.toList $ SU.fromList) . fmap (second Seq.singleton)+{-# INLINABLE groupByOrderedKey #-}++toSeq :: Monad m => SF.Fold m a (Seq.Seq a)+toSeq = SF.foldl' (\s a -> s <> Seq.singleton a) mempty+{-# INLINEABLE toSeq #-}++-- | Group streamly stream of @(k,c)@ by ordered key. Using toMapIO for mutable cells in the fold.+groupByOrderedKeyIO+  :: (Monad m, MonadIO m, Ord k) => S.Stream m (k, c) -> S.Stream m (k, Seq.Seq c)+groupByOrderedKeyIO = streamMeta (SF.toMapIO fst (SF.lmap snd toSeq)) (SU.lmap MS.toList $ SU.fromList)+{-# INLINABLE groupByOrderedKeyIO #-}++{-+toHashMapSeqST :: (Monad m, Eq k, Hashable k) => SF.Fold m (k, a) (ST.ST s (HTC.HashTable s k (Seq.Seq a)))+toHashMapSeqST = SF.foldl' (\hm (k, a) -> traverse (mutate k (f a)) hm) HTC.new where+  mutate k f hm = HTC.mutate hm k f+  f :: forall s . a -> Maybe (Seq.Seq a) -> Maybe (Seq.Seq a, ((HTC.HashTable s k (Seq.Seq a))))+  f a ht saM = case saM of+    Nothing -> (Seq.singleton a, ())+    Just sa -> (sa Seq.|> a, ())+{-# INLINEABLE toHashMapSeqST #-}+-}++-- | Group streamly stream of @(k,c)@ by @hashable@ key. Uses mutable hashtables running in the ST monad.+-- NB: this function uses the fact that @SerialT m@ is a monad+groupByHashableKeyST+  :: forall m k c . (Monad m, Hashable k, Eq k)+  => S.Stream m (k, c)+  -> S.Stream m (k, Seq.Seq c)+groupByHashableKeyST st = S.concatEffect $ fmap listToStream $ S.toList st+--  lkc <- fromEffect (S.toList st)+  where+    listToStream :: [(k, c)] -> S.Stream m (k, Seq.Seq c)+    listToStream l = S.fromList $ ST.runST $ listToList l+    listToList :: [(k, c)] -> ST.ST s [(k, Seq.Seq c)]+    listToList = join . fmap HT.toList . toHT+    toHT :: [(k, c)] -> ST.ST s (HTC.HashTable s k (Seq.Seq c))+    toHT l = (MRE.fromListWithHT @HTC.HashTable) (<>)+        $ fmap (second Seq.singleton) l+{-# INLINABLE groupByHashableKeyST #-}+++-- | Group streamly stream of @(k,c)@ by key with instance of Grouping from <http://hackage.haskell.org/package/discrimination>.+-- NB: this function uses the fact that @SerialT m@ is a monad+groupByDiscriminatedKey+  :: (Monad m, DG.Grouping k)+  => S.Stream m (k, c)+  -> S.Stream m (k, Seq.Seq c)+groupByDiscriminatedKey s =+  S.concatEffect+  $ (StreamK.toStream . StreamK.fromFoldable . Maybe.catMaybes . fmap (fmap g . LNE.nonEmpty) . DG.groupWith fst)+  <$> S.toList s+  where+    g :: LNE.NonEmpty (k, c) -> (k, Seq.Seq c)+    g x = let k = fst (LNE.head x) in (k, F.fold $ fmap (Seq.singleton . snd) x)+{-# INLINABLE groupByDiscriminatedKey #-}+#else+ unpackStream :: S.IsStream t => MRC.Unpack x y -> t Identity x -> t Identity y unpackStream (MRC.Filter t) = S.filter t unpackStream (MRC.Unpack f) = S.concatMap (S.fromFoldable . f)@@ -286,3 +493,4 @@       g x = let k = fst (LNE.head x) in (k, F.fold $ fmap (Seq.singleton . snd) x)   S.fromFoldable $ Maybe.catMaybes . fmap (fmap g . LNE.nonEmpty) $ DG.groupWith fst lkc {-# INLINABLE groupByDiscriminatedKey #-}+#endif