map-reduce-folds (empty) → 0.1.0.0
raw patch · 17 files changed
+2036/−0 lines, 17 filesdep +basedep +containersdep +criterion
Dependencies added: base, containers, criterion, deepseq, discrimination, dump-core, foldl, hashable, hashtables, hedgehog, map-reduce-folds, parallel, profunctors, random, split, streaming, streamly, text, unordered-containers, vector
Files
- ChangeLog.md +3/−0
- LICENSE +30/−0
- bench/MapReduce.hs +369/−0
- examples/ListStats.hs +81/−0
- examples/readmeExample.hs +28/−0
- map-reduce-folds.cabal +111/−0
- src/Control/MapReduce.hs +44/−0
- src/Control/MapReduce/Core.hs +260/−0
- src/Control/MapReduce/Engines.hs +108/−0
- src/Control/MapReduce/Engines/List.hs +104/−0
- src/Control/MapReduce/Engines/ParallelList.hs +85/−0
- src/Control/MapReduce/Engines/Streaming.hs +161/−0
- src/Control/MapReduce/Engines/Streamly.hs +237/−0
- src/Control/MapReduce/Engines/Vector.hs +117/−0
- src/Control/MapReduce/Simple.hs +222/−0
- test/Test1.hs +66/−0
- test/TestAll.hs +10/−0
+ ChangeLog.md view
@@ -0,0 +1,3 @@+version 0.1.0.0++initial Hackage version
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2018, Adam Conner-Sax++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * 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.++ * Neither the name of Adam Conner-Sax nor the names of other+ 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+OWNER 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.
+ bench/MapReduce.hs view
@@ -0,0 +1,369 @@+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+import Criterion.Main+import Criterion+++import Control.MapReduce as MR+import Control.MapReduce.Engines.List+ as MRL+import Control.MapReduce.Engines.Streaming+ as MRS+import Control.MapReduce.Engines.Streamly+ as MRSL+import Control.MapReduce.Engines.Vector+ as MRV+import Control.MapReduce.Engines.ParallelList+ as MRP+import Data.Functor.Identity ( runIdentity )+import Data.Text as T+import Data.List as L+import Data.HashMap.Lazy as HM+import Data.Map as M+import qualified Control.Foldl as FL+import Control.Arrow ( second )+import Data.Foldable as F+import Data.Functor.Identity ( Identity(Identity)+ , runIdentity+ )+import Data.Sequence as Seq+import Data.Maybe ( catMaybes )+import System.Random ( newStdGen+ , randomRs+ , randomRIO+ )+++createPairData :: Int -> IO [(Char, Int)]+createPairData n = do+ g <- newStdGen+ let randLabels = L.take n $ randomRs ('A', 'Z') g+ randInts = L.take n $ randomRs (1, 100) g+ return $ L.zip randLabels randInts++benchPure :: (NFData b) => String -> (Int -> a) -> (a -> b) -> Benchmark+benchPure name src f =+ bench name $ nfIO $ randomRIO (1, 1) >>= return . f . src++benchIO :: (NFData b) => String -> (Int -> a) -> (a -> IO b) -> Benchmark+benchIO name src f = bench name $ nfIO $ randomRIO (1, 1) >>= f . src++-- For example, keep only even numbers, then compute the average of the Int for each label.+filterPF = even . snd+assignPF = id -- this is a function from the "data" to a pair (key, data-to-process).+reducePFold = FL.premap realToFrac FL.mean+reducePF k = fmap (M.singleton k) $ reducePFold -- k -> Fold x (Map k x)+doReducePF k = FL.fold (reducePF k)++direct :: Foldable g => g (Char, Int) -> M.Map Char Double --[(Char, Double)]+direct =+ F.fold+ . fmap (uncurry doReducePF)+ . HM.toList+ . HM.fromListWith (<>)+ . fmap (second $ pure @[])+ . L.filter filterPF+ . F.toList+{-# INLINE direct #-}+++directFoldl :: Foldable g => g (Char, Int) -> M.Map Char Double+directFoldl =+ F.fold+ . fmap (uncurry doReducePF)+ . HM.toList+ . HM.fromListWith (<>)+ . fmap (second $ pure @[])+ . L.filter filterPF+ . FL.fold FL.list+{-# INLINE directFoldl #-}++++mapReduceList :: Foldable g => g (Char, Int) -> M.Map Char Double+mapReduceList = FL.fold+ (fmap+ F.fold+ (MRL.listEngine MRL.groupByHashableKey+ (MR.Filter filterPF)+ (MR.Assign id)+ (MR.ReduceFold reducePF)+ )+ )+{-# INLINE mapReduceList #-}++mapReduceListP :: Foldable g => g (Char, Int) -> M.Map Char Double+mapReduceListP = FL.fold+ (fmap+ F.fold+ (MRP.parallelListEngine 6+ MRL.groupByHashableKey+ (MR.Filter filterPF)+ (MR.Assign id)+ (MR.ReduceFold reducePF)+ )+ )+{-# INLINE mapReduceListP #-}+++mapReduceStreaming :: Foldable g => g (Char, Int) -> M.Map Char Double+mapReduceStreaming = FL.fold+ (MRS.concatStreamFold+ (MRS.streamingEngine MRS.groupByHashableKey+ (MR.Filter filterPF)+ (MR.Assign id)+ (MR.ReduceFold reducePF)+ )+ )+{-# INLINE mapReduceStreaming #-}++mapReduceStreamlyOrd :: Foldable g => g (Char, Int) -> M.Map Char Double+mapReduceStreamlyOrd = FL.fold+ (MRSL.concatStreamFold+ (MRSL.streamlyEngine MRSL.groupByOrderedKey+ (MR.Filter filterPF)+ (MR.Assign id)+ (MR.ReduceFold reducePF)+ )+ )+{-# INLINE mapReduceStreamlyOrd #-}++mapReduceStreamlyHash :: Foldable g => g (Char, Int) -> M.Map Char Double+mapReduceStreamlyHash = FL.fold+ (MRSL.concatStreamFold+ (MRSL.streamlyEngine MRSL.groupByHashableKey+ (MR.Filter filterPF)+ (MR.Assign id)+ (MR.ReduceFold reducePF)+ )+ )+{-# INLINE mapReduceStreamlyHash #-}++mapReduceStreamlyHashST :: Foldable g => g (Char, Int) -> M.Map Char Double+mapReduceStreamlyHashST = FL.fold+ (MRSL.concatStreamFold+ (MRSL.streamlyEngine MRSL.groupByHashableKeyST+ (MR.Filter filterPF)+ (MR.Assign id)+ (MR.ReduceFold reducePF)+ )+ )+{-# INLINE mapReduceStreamlyHashST #-}++mapReduceStreamlyDiscrimination+ :: Foldable g => g (Char, Int) -> M.Map Char Double+mapReduceStreamlyDiscrimination = FL.fold+ (MRSL.concatStreamFold+ (MRSL.streamlyEngine MRSL.groupByDiscriminatedKey+ (MR.Filter filterPF)+ (MR.Assign id)+ (MR.ReduceFold reducePF)+ )+ )+{-# INLINE mapReduceStreamlyDiscrimination #-}+++mapReduceStreamlyC+ :: forall tIn tOut m g+ . (MonadAsync m, Foldable g, MRSL.IsStream tIn, MRSL.IsStream tOut)+ => g (Char, Int)+ -> m (M.Map Char Double)+mapReduceStreamlyC = FL.foldM+ (MRSL.concatConcurrentStreamFold+ ((MRSL.concurrentStreamlyEngine @tIn @tOut) MRSL.groupByHashableKey+ (MR.Filter filterPF)+ (MR.Assign id)+ (MR.ReduceFold reducePF)+ )+ )+{-# INLINE mapReduceStreamlyC #-}+++mapReduceVector :: Foldable g => g (Char, Int) -> M.Map Char Double+mapReduceVector = FL.fold+ (fmap+ F.fold+ (MRV.vectorEngine MRV.groupByHashableKey+ (MR.Filter filterPF)+ (MR.Assign id)+ (MR.ReduceFold reducePF)+ )+ )+{-# INLINE mapReduceVector #-}++{-+parMapReduce :: Foldable g => g (Char, Int) -> [(Char, Double)]+parMapReduce = FL.fold+ (MRP.parallelMapReduceFold+ 6+ (MR.Unpack $ \x -> if filterPF x then [x] else [])+ (MR.Assign id)+ (MR.Reduce reducePF)+ )+{-# INLINE parMapReduce #-}+-}++++benchOne dat = bgroup+ "Task 1, on (Char, Int) "+ [ benchPure "direct" (const dat) direct+ , benchPure "directFoldl" (const dat) directFoldl+ , benchPure "mapReduce ([] Engine, strict hash map)" (const dat) mapReduceList+ , benchPure "mapReduce (Streaming.Stream Engine, strict hash map)"+ (const dat)+ mapReduceStreaming+ , benchPure "mapReduce (Streamly.SerialT Engine, strict map)"+ (const dat)+ mapReduceStreamlyOrd+ , benchPure "mapReduce (Streamly.SerialT Engine, strict hash map)"+ (const dat)+ mapReduceStreamlyHash+ , benchPure "mapReduce (Streamly.SerialT Engine, strict hash table, ST)"+ (const dat)+ mapReduceStreamlyHashST+ , benchPure "mapReduce (Streamly.SerialT Engine, discrimination)"+ (const dat)+ mapReduceStreamlyDiscrimination+ , benchPure "mapReduce (Data.Vector Engine, strict hash map)"+ (const dat)+ mapReduceVector+ ]++benchConcurrent dat = bgroup+ "Task 1, on (Char, Int). Concurrent Engines"+ [ benchPure "list, parallel (6 threads)" (const dat) mapReduceListP+ , benchIO "streamly, parallely"+ (const dat)+ (mapReduceStreamlyC @MRSL.SerialT @MRSL.ParallelT)+ , benchIO "streamly, aheadly"+ (const dat)+ (mapReduceStreamlyC @MRSL.SerialT @MRSL.AheadT)+ , benchIO "streamly, asyncly"+ (const dat)+ (mapReduceStreamlyC @MRSL.SerialT @MRSL.AsyncT)+ ]++-- a more complex row type+createMapRows :: Int -> IO (Seq.Seq (M.Map T.Text Int))+createMapRows n = do+ g <- newStdGen+ let randInts = L.take (n + 1) $ randomRs (1, 100) g+ makeRow k =+ let l = randInts !! k+ in if even l+ then M.fromList [("A", l), ("B", l `mod` 47), ("C", l `mod` 13)]+ else M.fromList [("A", l), ("B", l `mod` 47)]+ return+ $ Seq.unfoldr (\m -> if m > n then Nothing else Just (makeRow m, m + 1)) 0++-- unpack: if A and B and C are present, unpack to Just (A,B,C), otherwise Nothing+unpackMF :: M.Map T.Text Int -> Maybe (Int, Int, Int)+unpackMF m = do+ a <- M.lookup "A" m+ b <- M.lookup "B" m+ c <- M.lookup "C" m+ return (a, b, c)++-- group by the value of "C"+assignMF :: (Int, Int, Int) -> (Int, (Int, Int))+assignMF (a, b, c) = (c, (a, b))++-- 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 ++-- return [(C, <A+B>)]++directM :: Foldable g => g (M.Map T.Text Int) -> M.Map Int Double+directM =+ fmap (FL.fold reduceMFold)+ . M.fromListWith (<>)+ . fmap (second (pure @[]) . assignMF)+ . catMaybes+ . fmap unpackMF+ . F.toList++mapReduce2List :: Foldable g => g (M.Map T.Text Int) -> M.Map Int Double+mapReduce2List = FL.fold+ (fmap+ F.fold+ (MRL.listEngine MRL.groupByHashableKey+ (MR.Unpack unpackMF)+ (MR.Assign assignMF)+ (MR.foldAndLabel reduceMFold M.singleton)+ )+ )++mapReduce2Streaming :: Foldable g => g (M.Map T.Text Int) -> M.Map Int Double+mapReduce2Streaming = FL.fold+ (MRS.concatStreamFold+ (MRS.streamingEngine MRS.groupByHashableKey+ (MR.Unpack unpackMF)+ (MR.Assign assignMF)+ (MR.foldAndLabel reduceMFold M.singleton)+ )+ )++mapReduce2Streamly :: Foldable g => g (M.Map T.Text Int) -> M.Map Int Double+mapReduce2Streamly = FL.fold+ (MRSL.concatStreamFold+ (MRSL.streamlyEngine MRSL.groupByHashableKey+ (MR.Unpack unpackMF)+ (MR.Assign assignMF)+ (MR.foldAndLabel reduceMFold M.singleton)+ )+ )++mapReduce2Vector :: Foldable g => g (M.Map T.Text Int) -> M.Map Int Double+mapReduce2Vector = FL.fold+ (fmap+ F.fold+ (MRV.vectorEngine MRV.groupByHashableKey+ (MR.Unpack unpackMF)+ (MR.Assign assignMF)+ (MR.foldAndLabel reduceMFold M.singleton)+ )+ )++{-+basicListP :: Foldable g => g (M.Map T.Text Int) -> [(Int, Double)]+basicListP = FL.fold+ (MRP.parallelMapReduceFold 6+ (MR.Unpack unpackMF)+ (MR.Assign assignMF)+ (MR.foldAndRelabel reduceMFold (\k x -> (k, x)))+ )+-}++benchTwo dat = bgroup+ "Task 2, on Map Text Int "+ [ benchPure "direct" (const dat) directM+ , benchPure "map-reduce-fold ([] Engine, strict hash map, serial)"+ (const dat)+ mapReduce2List+ , benchPure+ "map-reduce-fold (Streaming.Stream Engine, strict hash map, serial)"+ (const dat)+ mapReduce2Streaming+ , benchPure+ "map-reduce-fold (Streamly.SerialT Engine, strict hash map, serial)"+ (const dat)+ mapReduce2Streamly+ , benchPure "map-reduce-fold (Data.Vector Engine, strict hash map, serial)"+ (const dat)+ mapReduce2Vector+ ]++main :: IO ()+main = do+ dat <- createPairData 100000+ dat2 <- createMapRows 100000+ defaultMain [benchOne dat, benchConcurrent dat, benchTwo dat2]
+ examples/ListStats.hs view
@@ -0,0 +1,81 @@+module Main where++import qualified Control.MapReduce.Simple as MR+import qualified Control.Foldl as FL++ints n = take n [1 ..]++multipleOf n x = (x `mod` n) == 0++onlyEven = MR.filterUnpack even -- a filter++andTwice x = [x, 2 * x] -- a "melt"++withTwice :: MR.Unpack Int Int+withTwice = MR.Unpack andTwice++groupByMultOf3 = MR.assign (multipleOf 3) realToFrac++-- An example using the function version+addAll :: (Foldable f, Num a) => f a -> a+addAll = FL.fold FL.sum+++reduceSum = MR.foldAndLabel FL.sum ((,))+reduceSum' = MR.processAndLabel addAll (,)+reduceMean = MR.foldAndLabel FL.mean ((,))++-- sum of all even ints, grouped by whether or not they are mutliples of 3+sumsF = MR.mapReduceFold onlyEven groupByMultOf3 reduceSum+sumsF' = MR.mapReduceFold onlyEven groupByMultOf3 reduceSum'++-- mean of all even ints, grouped by whether or not they are mutliples of 3+meansF = MR.mapReduceFold onlyEven groupByMultOf3 reduceMean++-- loop over initial list once but group each time+sumsAndMeansF = (,) <$> sumsF <*> meansF++-- loop over initial list once and group only once.+sumAndMeanEachF =+ MR.mapReduceFold onlyEven groupByMultOf3 ((,) <$> reduceSum <*> reduceMean)++-- sum and mean of all ints--and twice each int--grouped by whether or not they are multiples of 3.+sumAndMeanWithTwiceF =+ MR.mapReduceFold withTwice groupByMultOf3 ((,) <$> reduceSum <*> reduceMean)+++{-+We combine all the folds we want to do into one fold. So we loop over the input list only once here.+We group it for each fold. But the last two, sumAndMeanEach and sumAndMeanWithTwice shows how we can avoid that as well.+-}++main :: IO ()+main = do+ let (s, s', m, sm, sme, smd) = MR.fold+ ( (,,,,,)+ <$> sumsF+ <*> sumsF'+ <*> meansF+ <*> sumsAndMeansF+ <*> sumAndMeanEachF+ <*> sumAndMeanWithTwiceF+ )+ (ints 100)+ putStrLn $ "Sums: " ++ show s+ putStrLn $ "Sums': " ++ show s'+ putStrLn $ "Means: " ++ show m+ putStrLn $ "Sums & Means: " ++ show sm+ putStrLn $ "Sum & Mean of each: " ++ show sme+ putStrLn $ "Sum & Mean of all plus doubles: " ++ show smd+++{- Output:++Sums: [(False,1734.0),(True,816.0)]+Sums': [(False,1734.0),(True,816.0)]+Means: [(False,51.0),(True,51.0)]+Sums & Means: ([(False,1734.0),(True,816.0)],[(False,51.0),(True,51.0)])+Sum & Mean of each: [((False,1734.0),(False,51.0)),((True,816.0),(True,51.0))]+Sum & Mean of all plus doubles: [((False,10101.0),(False,75.38059701492539)),((True,5049.0),(True,76.5))]++-}
+ examples/readmeExample.hs view
@@ -0,0 +1,28 @@+import qualified Control.MapReduce.Simple as MR+import qualified Control.MapReduce.Engines.Streamly+ as MRS+import qualified Control.Foldl as FL+import qualified Data.Map as M++unpack = MR.filterUnpack even -- filter, keeping even numbers++-- assign each number a key of True or False depending on whether it's a multiple of 3. Group the number itself.+isMultipleOf3 x = (x `mod` 3) == 0+assign = MR.assign isMultipleOf3 id++reduce = MR.foldAndLabel FL.sum M.singleton -- sum each group and then create a singleton Map for later combining++-- of the first n integers, compute the sum of even numbers which are mutliples of 3 +-- and the sum of all the even numbers which are not multiples of 3. +mrFold :: FL.Fold Int (M.Map Bool Int)+mrFold = MRS.concatStreamFold+ $ MRS.streamlyEngine MRS.groupByHashableKey unpack assign reduce++result :: Int -> M.Map Bool Int+result n = FL.fold mrFold [1 .. n]++main = putStrLn $ show $ result 10++{- Output+fromList [(False,24),(True,6)]+-}
+ map-reduce-folds.cabal view
@@ -0,0 +1,111 @@+cabal-version: 2.2+ +name: map-reduce-folds+version: 0.1.0.0+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.+bug-reports: https://github.com/adamConnerSax/map-reduce-folds/issues+license: BSD-3-Clause+license-file: LICENSE+author: Adam Conner-Sax+maintainer: adam_conner_sax@yahoo.com+copyright: 2019 Adam Conner-Sax+category: Control+extra-source-files: ChangeLog.md+tested-with: GHC ==8.6.4 || ==8.6.2+flag dump-core+ description: Dump HTML for the core generated by GHC during compilation+ default: False+++source-repository head+ Type: git+ Location: https://github.com/adamConnerSax/map-reduce-folds+ +common deps+ build-depends: base >= 4.12.0 && < 4.13,+ containers >= 0.6.0 && < 0.7,+ foldl >= 1.4.5 && < 1.5,+ profunctors >= 5.3 && < 5.4,+ text >= 1.2.3 && < 1.3,+ unordered-containers >= 0.2.10 && < 0.3+ +library+ import: deps+ if flag(dump-core)+ build-depends: dump-core+ ghc-options: -fforce-recomp -fplugin=DumpCore -fplugin-opt DumpCore:core-html + ghc-options: -Wall -fspecialise-aggressively -funbox-strict-fields -fexpose-all-unfoldings+ exposed-modules: Control.MapReduce+ , Control.MapReduce.Core+ , Control.MapReduce.Simple+ , Control.MapReduce.Engines+ , Control.MapReduce.Engines.Streaming+ , Control.MapReduce.Engines.Streamly+ , Control.MapReduce.Engines.Vector+ , Control.MapReduce.Engines.List+ , Control.MapReduce.Engines.ParallelList++ build-depends:+ discrimination >= 0.3 && < 0.4,+ hashable >= 1.2.7 && < 1.3,+ hashtables >= 1.2.0.0 && < 1.3.0.0,+ vector >= 0.12.0 && < 0.13,+ parallel >= 3.2.2 && < 3.3,+ split >= 0.2.3 && < 0.3,+ streaming >= 0.2.2 && < 0.3,+ streamly >= 0.6.1 && < 0.7++ hs-source-dirs: src+ default-language: Haskell2010+++test-suite map-reduce-folds-test+ type: exitcode-stdio-1.0+ main-is: TestAll.hs+ other-modules: Test1+ hs-source-dirs: test+ ghc-options: + build-depends:+ base+ , hedgehog >= 0.6.0 && < 0.7+ , map-reduce-folds + , containers+ , foldl+ + default-language: Haskell2010+ +benchmark bench-map-reduce+ import: deps+ if flag(dump-core)+ build-depends: dump-core+ ghc-options: -fforce-recomp -fplugin=DumpCore -fplugin-opt DumpCore:core-html + type: exitcode-stdio-1.0+ hs-source-dirs: bench+ other-modules: + ghc-options: -O3 -fspecialise-aggressively -funbox-strict-fields -fexpose-all-unfoldings -fforce-recomp+-- -threaded "-with-rtsopts=-N"+ main-is: MapReduce.hs+ build-depends: map-reduce-folds >= 0.1.0.0, + criterion ,+ deepseq ,+ random+ default-language: Haskell2010+++executable listStats+ import: deps+ main-is: ListStats.hs+ hs-source-dirs: examples+ ghc-options: -Wall+ build-depends: map-reduce-folds -any+ default-language: Haskell2010++executable readmeExample+ import: deps+ main-is: readmeExample.hs+ hs-source-dirs: examples+ ghc-options: -Wall+ build-depends: map-reduce-folds -any+ default-language: Haskell2010
+ src/Control/MapReduce.hs view
@@ -0,0 +1,44 @@+{-|+Module : Control.MapReduce+Description : a map-reduce wrapper around foldl +Copyright : (c) Adam Conner-Sax 2019+License : BSD-3-Clause+Maintainer : adam_conner_sax@yahoo.com+Stability : experimental++MapReduce as folds+This is all just wrapping around Control.Foldl so that it's easier to see the map-reduce structure+The Mapping step is broken into 2 parts:++1. unpacking, which could include "melting" or filtering,++2. assigning, which assigns a group to each unpacked item. Could just be choosing a key column(s)++The items are then grouped by key and "reduced"++The reduce step is conceptually simpler, just requiring a function from the (key, grouped data) pair to the result.++Reduce could be as simple as combining the key with a single data row or some very complex function of the grouped data.+E.g., reduce could itself be a map-reduce on the grouped data.+Since these are folds, we can share work by using the Applicative instance of MapStep (just the Applicative instance of Control.Foldl.Fold)+and we will loop over the data only once.+The Reduce type is also Applicative so there could be work sharing there as well:+e.g., if your `reduce :: (k -> d -> e)` has the form `reduce k :: FL.Fold d e`++These types are meant to simplify the building of "Engines" which combine them into a single efficient fold from a container of x to some container of the result. The Engine amounts to a choice of grouping algorithm (usually just uses Data.Map or Data.HashMap) and a type in which to do the calculations and return the result. These are assembled into a Fold.++The goal is to make assembling a large family of common map/reduce patterns in a straightforward way. At some level of complication, you may as+well write them by hand. An in-between case would be writing the unpack function as a complex hand written filter+-}+module Control.MapReduce+ (+ -- * Core types for specifying map/reduce folds+ module Control.MapReduce.Core+ -- * default choices for grouping algorithms and intermediate types and helper functions+ -- for common cases.+ , module Control.MapReduce.Simple+ )+where+import Control.MapReduce.Core+import Control.MapReduce.Simple+
+ src/Control/MapReduce/Core.hs view
@@ -0,0 +1,260 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}+{-|+Module : Control.MapReduce.Core+Description : a map-reduce wrapper around foldl +Copyright : (c) Adam Conner-Sax 2019+License : BSD-3-Clause+Maintainer : adam_conner_sax@yahoo.com+Stability : experimental+++MapReduce as folds+This is all just wrapping around Control.Foldl so that it's easier to see the map-reduce structure+The Mapping step is broken into 2 parts:++1. unpacking, which could include "melting" or filtering,++2. assigning, which assigns a group to each unpacked item. Could just be choosing a key column(s)++The items are then grouped by key and "reduced"++The reduce step is conceptually simpler, just requiring a function from the (key, grouped data) pair to the result.++Reduce could be as simple as combining the key with a single data row or some very complex function of the grouped data.+E.g., reduce could itself be a map-reduce on the grouped data.+Since these are folds, we can share work by using the Applicative instance of MapStep (just the Applicative instance of Control.Foldl.Fold)+and we will loop over the data only once.+The Reduce type is also Applicative so there could be work sharing there as well, especially if you+specify your reduce as a Fold.+e.g., if your @reduce :: (k -> h c -> d)@ has the form @reduce :: k -> FL.Fold c d@++We combine these steps with an "Engine", resulting in a fold from a container of @x@ to some container of @d@.+The Engine amounts to a choice of grouping algorithm (usually using @Data.Map@ or @Data.HashMap@) and a choice of result container type.+The result container type is used for the intermediate steps as well.++The goal is to make assembling a large family of common map/reduce patterns in a straightforward way. At some level of complication, you may as+well write them by hand. An in-between case would be writing the unpack function as a complex hand written filter+-}+module Control.MapReduce.Core+ (+ -- * Basic Types for map reduce+ -- ** non-monadic+ Unpack(..)+ , Assign(..)+ , Reduce(..)++ -- ** monadic+ , UnpackM(..)+ , AssignM(..)+ , ReduceM(..)++ -- ** functions to generalize non-monadic to monadic+ , generalizeUnpack+ , generalizeAssign+ , generalizeReduce++ -- * Foldl helpers + , functionToFold+ , functionToFoldM+ , postMapM++ -- * re-exports+ , Fold+ , FoldM+ , fold+ , foldM+ )+where++import qualified Control.Foldl as FL+import Control.Foldl ( Fold+ , FoldM+ , fold+ , foldM+ ) -- for re-exporting++import qualified Data.Profunctor as P+import Data.Profunctor ( Profunctor )+import qualified Data.Sequence as S+import Control.Arrow ( second )++-- | @Unpack@ is for filtering rows or "melting" them, that is doing something which turns each @x@ into several @y@.+-- Filter is a special case because it can often be done faster directly than via folding over @Maybe@+data Unpack x y where+ Filter :: (x -> Bool) -> Unpack x x -- we single out this special case because it's faster to do directly+ Unpack :: Traversable g => (x -> g y) -> Unpack x y -- we only need (Functor g, Foldable g) but if we want to generalize we need Traversable++-- helpers for turning filter functions into @a -> Maybe a@+boolToMaybe :: Bool -> a -> Maybe a+boolToMaybe b x = if b then Just x else Nothing++ifToMaybe :: (x -> Bool) -> x -> Maybe x+ifToMaybe t x = boolToMaybe (t x) x++instance Functor (Unpack x) where+ fmap h (Filter t) = Unpack (fmap h . ifToMaybe t)+ fmap h (Unpack f) = Unpack (fmap h . f)+ {-# INLINABLE fmap #-}++instance P.Profunctor Unpack where+ dimap l r (Filter t) = Unpack ( fmap r . ifToMaybe t . l)+ dimap l r (Unpack f) = Unpack ( fmap r . f . l)+ {-# INLINABLE dimap #-}++-- | @UnpackM@ is for filtering or melting the input type. This version has a monadic result type to+-- accomodate unpacking that might require randomness or logging during unpacking.+-- filter is a special case since filtering can be often be done faster. So we single it out. +data UnpackM m x y where+ FilterM :: Monad m => (x -> m Bool) -> UnpackM m x x+ UnpackM :: (Monad m, Traversable g) => (x -> m (g y)) -> UnpackM m x y++ifToMaybeM :: Monad m => (x -> m Bool) -> x -> m (Maybe x)+ifToMaybeM t x = fmap (`boolToMaybe` x) (t x)++instance Functor (UnpackM m x) where+ fmap h (FilterM t) = UnpackM (fmap (fmap h) . ifToMaybeM t)+ fmap h (UnpackM f) = UnpackM (fmap (fmap h) . f)+ {-# INLINABLE fmap #-}++instance Profunctor (UnpackM m) where+ dimap l r (FilterM t) = UnpackM ( fmap (fmap r) . ifToMaybeM t . l)+ dimap l r (UnpackM f) = UnpackM ( fmap (fmap r) . f . l)+ {-# INLINABLE dimap #-}++-- | lift a non-monadic Unpack to a monadic one for any monad m+generalizeUnpack :: Monad m => Unpack x y -> UnpackM m x y+generalizeUnpack (Filter t) = FilterM $ return . t+generalizeUnpack (Unpack f) = UnpackM $ return . f+{-# INLINABLE generalizeUnpack #-}++-- | map @y@ into a @(k,c)@ pair for grouping +data Assign k y c where+ Assign :: (y -> (k, c)) -> Assign k y c++instance Functor (Assign k y) where+ fmap f (Assign h) = Assign $ second f . h --(\y -> let (k,c) = g y in (k, f c))+ {-# INLINABLE fmap #-}++instance Profunctor (Assign k) where+ dimap l r (Assign h) = Assign $ second r . h . l --(\z -> let (k,c) = g (l z) in (k, r c))+ {-# INLINABLE dimap #-}++-- | Effectfully map @y@ into a @(k,c)@ pair for grouping +data AssignM m k y c where+ AssignM :: Monad m => (y -> m (k, c)) -> AssignM m k y c++instance Functor (AssignM m k y) where+ fmap f (AssignM h) = AssignM $ fmap (second f) . h+ {-# INLINABLE fmap #-}++instance Profunctor (AssignM m k) where+ dimap l r (AssignM h) = AssignM $ fmap (second r) . h . l+ {-# INLINABLE dimap #-}+++-- | lift a non-monadic Assign to a monadic one for any monad m+generalizeAssign :: Monad m => Assign k y c -> AssignM m k y c+generalizeAssign (Assign h) = AssignM $ return . h+{-# INLINABLE generalizeAssign #-}++-- | Wrapper for functions to reduce keyed and grouped data to the result type.+-- It is *strongly* suggested that you use Folds for this step. Type-inference and+-- applicative optimization are more straightforward that way. The non-Fold contructors are+-- there in order to retro-fit existing functions.++-- | Reduce step for non-effectful reductions+data Reduce k x d where+ Reduce :: (k -> (forall h. (Foldable h, Functor h) => (h x -> d))) -> Reduce k x d+ ReduceFold :: (k -> FL.Fold x d) -> Reduce k x d++-- | Reduce step for effectful reductions.+-- It is *strongly* suggested that you use Folds for this step. Type-inference and+-- applicative optimization are more straightforward that way. The non-Fold contructors are+-- there in order to retro-fit existing functions.+data ReduceM m k x d where+ ReduceM :: Monad m => (k -> (forall h. (Foldable h, Functor h) => (h x -> m d))) -> ReduceM m k x d+ ReduceFoldM :: Monad m => (k -> FL.FoldM m x d) -> ReduceM m k x d++instance Functor (Reduce k x) where+ fmap f (Reduce g) = Reduce $ \k -> f . g k+ fmap f (ReduceFold g) = ReduceFold $ \k -> fmap f (g k)+ {-# INLINABLE fmap #-}++instance Functor (ReduceM m k x) where+ fmap f (ReduceM g) = ReduceM $ \k -> fmap f . g k+ fmap f (ReduceFoldM g) = ReduceFoldM $ \k -> fmap f (g k)+ {-# INLINABLE fmap #-}++instance Profunctor (Reduce k) where+ dimap l r (Reduce g) = Reduce $ \k -> P.dimap (fmap l) r (g k)+ dimap l r (ReduceFold g) = ReduceFold $ \k -> P.dimap l r (g k)+ {-# INLINABLE dimap #-}++instance Profunctor (ReduceM m k) where+ dimap l r (ReduceM g) = ReduceM $ \k -> P.dimap (fmap l) (fmap r) (g k)+ dimap l r (ReduceFoldM g) = ReduceFoldM $ \k -> P.dimap l r (g k)+ {-# INLINABLE dimap #-}++instance Applicative (Reduce k x) where+ pure x = ReduceFold $ const (pure x)+ {-# INLINABLE pure #-}+ Reduce r1 <*> Reduce r2 = Reduce $ \k -> r1 k <*> r2 k+ ReduceFold f1 <*> ReduceFold f2 = ReduceFold $ \k -> f1 k <*> f2 k+ Reduce r1 <*> ReduceFold f2 = Reduce $ \k -> r1 k <*> FL.fold (f2 k)+ ReduceFold f1 <*> Reduce r2 = Reduce $ \k -> FL.fold (f1 k) <*> r2 k+ {-# INLINABLE (<*>) #-}++instance Monad m => Applicative (ReduceM m k x) where+ pure x = ReduceM $ \_ -> pure $ pure x+ {-# INLINABLE pure #-}+ ReduceM r1 <*> ReduceM r2 = ReduceM $ \k -> (<*>) <$> r1 k <*> r2 k+ ReduceFoldM f1 <*> ReduceFoldM f2 = ReduceFoldM $ \k -> f1 k <*> f2 k+ ReduceM r1 <*> ReduceFoldM f2 = ReduceM $ \k -> (<*>) <$> r1 k <*> FL.foldM (f2 k)+ ReduceFoldM f1 <*> ReduceM r2 = ReduceM $ \k -> (<*>) <$> FL.foldM (f1 k) <*> r2 k+ {-# INLINABLE (<*>) #-}++-- | Make a non-monadic reduce monadic. +generalizeReduce :: Monad m => Reduce k x d -> ReduceM m k x d+generalizeReduce (Reduce f) = ReduceM $ \k -> return . f k+generalizeReduce (ReduceFold f) = ReduceFoldM $ \k -> FL.generalize (f k)+{-# INLINABLE generalizeReduce #-}++-- TODO: submit a PR to foldl for this+-- | Given an effectful (monadic) Control.Foldl fold, a @FoldM m x a@, we can use its @Functor@ instance+-- to apply a non-effectful @(a -> b)@ to its result type. To apply @(a -> m b)@, we need this combinator.+postMapM :: Monad m => (a -> m b) -> FL.FoldM m x a -> FL.FoldM m x b+postMapM f (FL.FoldM step begin done) = FL.FoldM step begin done'+ where done' x = done x >>= f+{-# INLINABLE postMapM #-}++seqF :: FL.Fold a (S.Seq a)+seqF = FL.Fold (S.|>) S.empty id++-- | convert a function which takes a foldable container of x and produces an a into a Fold x a.+-- uses Data.Sequence.Seq as an intermediate type because it should behave well whether+-- f is a left or right fold. +-- This can be helpful in putting an existing function into a Reduce.+functionToFold :: (forall h . Foldable h => h x -> a) -> FL.Fold x a+functionToFold f = fmap f seqF++-- | convert a function which takes a foldable container of x and produces an (m a) into a FoldM m x a.+-- uses Data.Sequence.Seq as an intermediate type because it should behave well whether+-- f is a left or right fold.+-- This can be helpful in putting an existing function into a ReduceM.+functionToFoldM+ :: Monad m => (forall h . Foldable h => h x -> m a) -> FL.FoldM m x a+functionToFoldM f = postMapM f $ FL.generalize seqF+
+ src/Control/MapReduce/Engines.hs view
@@ -0,0 +1,108 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}+{-|+Module : Control.MapReduce.Engines+Description : map-reduce-folds builders+Copyright : (c) Adam Conner-Sax 2019+License : BSD-3-Clause+Maintainer : adam_conner_sax@yahoo.com+Stability : experimental++Types and functions used by all the engines.+Notes:++ 1. The provided grouping functions group elements into a 'Data.Sequence.Seq' as this is a good default choice.+ 2. The <http://hackage.haskell.org/package/streamly Streamly> engine is the fastest in my benchmarks. It's the engine used by default if you import @Control.MapReduce.Simple@.+ 3. All the engines take a grouping function as a parameter and default ones are provided. For simple map/reduce, the grouping step may be the bottleneck and I wanted to leave room for experimentation. I've tried (and failed!) to find anything faster than using 'Map' or 'HashMap' via @toList . fromListWith (<>)@.++-}+module Control.MapReduce.Engines+ (+ -- * Fold Types+ MapReduceFold+ , MapReduceFoldM++ -- * Engine Helpers+ , reduceFunction+ , reduceFunctionM++ -- * @groupBy@ Helpers+ , fromListWithHT+ )+where++import qualified Control.MapReduce.Core as MRC+import qualified Control.Foldl as FL+import Control.Monad.ST as ST+import Data.Hashable ( Hashable )++import qualified Data.HashTable.Class as HT+++-- | Type-alias for a map-reduce-fold engine+type MapReduceFold y k c q x d = MRC.Unpack x y -> MRC.Assign k y c -> MRC.Reduce k c d -> FL.Fold x (q d)++-- | Type-alias for a monadic (effectful) map-reduce-fold engine+type MapReduceFoldM m y k c q x d = MRC.UnpackM m x y -> MRC.AssignM m k y c -> MRC.ReduceM m k c d -> FL.FoldM m x (q d)++-- | Turn @Reduce@ into a function we can apply+reduceFunction :: (Foldable h, Functor h) => MRC.Reduce k x d -> k -> h x -> d+reduceFunction (MRC.Reduce f) k = f k+reduceFunction (MRC.ReduceFold f) k = FL.fold (f k)+{-# INLINABLE reduceFunction #-}++-- | Turn @ReduceM@ into a function we can apply+reduceFunctionM+ :: (Traversable h, Monad m) => MRC.ReduceM m k x d -> k -> h x -> m d+reduceFunctionM (MRC.ReduceM f) k = f k+reduceFunctionM (MRC.ReduceFoldM f) k = FL.foldM (f k)+{-# INLINABLE reduceFunctionM #-}++{-+-- copied from Frames+-- which causes overlapping instances. +instance {-# OVERLAPPABLE #-} Grouping Text where+ grouping = contramap hash grouping+-}++{- | an implementation of @fromListWith@ for mutable hashtables from the <http://hackage.haskell.org/package/hashtables-1.2.3.1 hastables>+package. Basically a copy @fromList@ from that package using mutate instead of insert to apply the given function if the+was already in the map. Might not be the ideal implementation.+Notes:++* This function is specific hashtable agnostic so you'll have to supply a specific implementation from the package via TypeApplication+* This function returns the hash-table in the @ST@ monad. You can fold over it (using @foldM@ from @hashtables@)+and then use @runST@ to get the grouped structure out.+-}+fromListWithHT+ :: forall h k v s+ . (HT.HashTable h, Eq k, Hashable k)+ => (v -> v -> v)+ -> [(k, v)]+ -> ST.ST s (h s k v)+fromListWithHT f l = do+ ht <- HT.new+ go ht l+ where+ g x mx = (Just $ maybe x (`f` x) mx, ())+ go ht = go'+ where+ go' [] = return ht+ go' ((!k, !v) : xs) = do+ HT.mutate ht k (g v)+ go' xs+{-# INLINABLE fromListWithHT #-}+
+ src/Control/MapReduce/Engines/List.hs view
@@ -0,0 +1,104 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE TypeApplications #-}+{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}+{-|+Module : Control.MapReduce.Engines.List+Description : map-reduce-folds builders+Copyright : (c) Adam Conner-Sax 2019+License : BSD-3-Clause+Maintainer : adam_conner_sax@yahoo.com+Stability : experimental++map-reduce engine (fold builder) using [] as its intermediate type.+-}+module Control.MapReduce.Engines.List+ (+ -- * Engines+ listEngine+ , listEngineM++ -- * @groupBy@ Functions+ , groupByHashableKey+ , groupByOrderedKey++ -- * Helpers+ , unpackList+ , unpackListM+ )+where++import qualified Control.MapReduce.Core as MRC+import qualified Control.MapReduce.Engines as MRE++import qualified Control.Foldl as FL+import qualified Data.List as L+import qualified Data.Foldable as F+import Data.Hashable ( Hashable )+import qualified Data.HashMap.Strict as HMS+import qualified Data.Map.Strict as MS+import qualified Data.Sequence as Seq+import Control.Monad ( filterM )+import Control.Arrow ( second )++++-- | unpack for list based map/reduce+unpackList :: MRC.Unpack x y -> [x] -> [y]+unpackList (MRC.Filter t) = L.filter t+unpackList (MRC.Unpack f) = L.concatMap (F.toList . f)+{-# INLINABLE unpackList #-}++-- | effectful unpack for list based map/reduce+unpackListM :: MRC.UnpackM m x y -> [x] -> m [y]+unpackListM (MRC.FilterM t) = filterM t+unpackListM (MRC.UnpackM f) = fmap L.concat . traverse (fmap F.toList . f)+{-# INLINABLE unpackListM #-}++-- | group the mapped and assigned values by key using a Data.HashMap.Strict+groupByHashableKey :: (Hashable k, Eq k) => [(k, c)] -> [(k, Seq.Seq c)]+groupByHashableKey =+ HMS.toList . HMS.fromListWith (<>) . fmap (second Seq.singleton)+{-# INLINABLE groupByHashableKey #-}++-- | group the mapped and assigned values by key using a Data.HashMap.Strict+groupByOrderedKey :: Ord k => [(k, c)] -> [(k, Seq.Seq c)]+groupByOrderedKey =+ MS.toList . MS.fromListWith (<>) . fmap (second Seq.singleton)+{-# INLINABLE groupByOrderedKey #-}++-- | map-reduce-fold builder using (Hashable k, Eq k) keys and returning a [] result+listEngine+ :: (Foldable g, Functor g)+ => ([(k, c)] -> [(k, g c)])+ -> MRE.MapReduceFold y k c [] x d+listEngine groupByKey u (MRC.Assign a) r = fmap+ (fmap (uncurry $ MRE.reduceFunction r) . groupByKey . fmap a . unpackList u)+ FL.list+{-# INLINABLE listEngine #-}++-- | effectful map-reduce-fold builder using (Hashable k, Eq k) keys and returning a [] result+listEngineM+ :: (Monad m, Traversable g)+ => ([(k, c)] -> [(k, g c)])+ -> MRE.MapReduceFoldM m y k c [] x d+listEngineM groupByKey u (MRC.AssignM a) rM = MRC.postMapM+ ( (traverse (uncurry $ MRE.reduceFunctionM rM) =<<)+ . fmap groupByKey+ . (traverse a =<<)+ . unpackListM u+ )+ (FL.generalize FL.list)+{-# INLINABLE listEngineM #-}+
+ src/Control/MapReduce/Engines/ParallelList.hs view
@@ -0,0 +1,85 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE PolyKinds #-}++{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE InstanceSigs #-}+{-# LANGUAGE TypeApplications #-}+{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}+{-|+Module : Control.MapReduce.Engines.ParallelList+Description : Gatherer code for map-reduce-folds+Copyright : (c) Adam Conner-Sax 2019+License : BSD-3-Clause+Maintainer : adam_conner_sax@yahoo.com+Stability : experimental++Some basic attempts at parallel map-reduce folds with lists as intermediate type. There are multiple places we can do things in parallel.++(1) We can map (unpack/assign) in parallel.+(2) We can reduce in parallel.+(3) We can fold our intermediate monoid (in the gatherer) in parallel+(4) We can fold our result monoid in parallel++NB: This does not seem to be faster--and is often slower!--than the serial engine. I leave it here as a starting point for improvement.+-}+module Control.MapReduce.Engines.ParallelList+ (+ -- * Parallel map/reduce fold builder+ parallelListEngine++ -- * re-exports+ , NFData+ )+where++import qualified Control.MapReduce.Core as MRC+import qualified Control.MapReduce.Engines as MRE+import qualified Control.MapReduce.Engines.List+ as MRL++import qualified Control.Foldl as FL+import qualified Data.List as L+import qualified Data.List.Split as L++import qualified Control.Parallel.Strategies as PS+import Control.Parallel.Strategies ( NFData ) -- for re-export+--+++{- | Parallel map-reduce-fold list engine. Uses the given parameters to use multiple sparks when mapping and reducing.+ Chunks the input to numThreads chunks and sparks each chunk for mapping, merges the results, groups, then uses the same chunking and merging to do the reductions.+ grouping could also be parallel but that is under the control of the given function.+-}+parallelListEngine+ :: forall g y k c x d+ . (NFData k, NFData c, NFData d, Foldable g, Functor g)+ => Int+ -> ([(k, c)] -> [(k, g c)])+ -> MRE.MapReduceFold y k c [] x d+parallelListEngine numThreads groupByKey u (MRC.Assign a) r =+ let chunkedF :: FL.Fold x [[x]] =+ fmap (L.divvy numThreads numThreads) FL.list+ mappedF :: FL.Fold x [(k, c)] =+ fmap (L.concat . parMapEach (fmap a . MRL.unpackList u)) chunkedF+ groupedF :: FL.Fold x [(k, g c)] = fmap groupByKey mappedF+ reducedF :: FL.Fold x [d] = fmap+ ( L.concat+ . parMapEach (fmap (uncurry $ MRE.reduceFunction r))+ . L.divvy numThreads numThreads+ )+ groupedF+ in reducedF++parMapEach :: PS.NFData b => (a -> b) -> [a] -> [b]+parMapEach = PS.parMap (PS.rparWith PS.rdeepseq)+{-# INLINABLE parMapEach #-}
+ src/Control/MapReduce/Engines/Streaming.hs view
@@ -0,0 +1,161 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE TypeApplications #-}+{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}+{-|+Module : Control.MapReduce.Engines.Streams+Description : map-reduce-folds builders+Copyright : (c) Adam Conner-Sax 2019+License : BSD-3-Clause+Maintainer : adam_conner_sax@yahoo.com+Stability : experimental++map-reduce engine (fold builder) using @Streaming.Streams@ as its intermediate type. Because @Streaming.Stream@ does not end with the+type of data data in the @Stream@, we wrap this type in @StreamResult@ for the purposes of the output type of the fold.+-}+module Control.MapReduce.Engines.Streaming+ (+ -- * Helper Types+ StreamResult(..)++ -- * Engines+ , streamingEngine+ , streamingEngineM++ -- * Result Extractors+ , resultToList+ , concatStream+ , concatStreamFold+ , concatStreamFoldM++ -- * @groupBy@ functions+ , groupByHashableKey+ , groupByOrderedKey+ )+where++import qualified Control.MapReduce.Core as MRC+import qualified Control.MapReduce.Engines as MRE++import qualified Control.Foldl as FL+import Data.Functor.Identity ( Identity )+import Data.Hashable ( Hashable )+import qualified Data.HashMap.Strict as HMS+import qualified Data.Map.Strict as MS+import qualified Data.Sequence as Seq+import qualified Streaming.Prelude as S+import qualified Streaming as S+import Streaming ( Stream+ , Of+ )+import Control.Arrow ( second )+++-- | unpack for streaming based map/reduce+unpackStream+ :: MRC.Unpack x y -> S.Stream (Of x) Identity r -> Stream (Of y) Identity r+unpackStream (MRC.Filter t) = S.filter t+unpackStream (MRC.Unpack f) = S.concat . S.map f+{-# INLINABLE unpackStream #-}++-- | effectful (monadic) unpack for streaming based map/reduce+unpackStreamM+ :: Monad m => MRC.UnpackM m x y -> Stream (Of x) m r -> Stream (Of y) m r+unpackStreamM (MRC.FilterM t) = S.filterM t+unpackStreamM (MRC.UnpackM f) = S.concat . S.mapM f+{-# INLINABLE unpackStreamM #-}++-- | group the mapped and assigned values by key using a @Data.HashMap.Strict@+groupByHashableKey+ :: forall m k c r+ . (Monad m, Hashable k, Eq k)+ => Stream (Of (k, c)) m r+ -> Stream (Of (k, Seq.Seq c)) m r+groupByHashableKey s = S.effect $ do+ (lkc S.:> r) <- S.toList s+ let hm = HMS.fromListWith (<>) $ fmap (second Seq.singleton) lkc+ return $ HMS.foldrWithKey (\k lc s' -> S.cons (k, lc) s') (return r) hm+{-# INLINABLE groupByHashableKey #-}++-- | group the mapped and assigned values by key using a @Data.Map.Strict@+groupByOrderedKey+ :: forall m k c r+ . (Monad m, Ord k)+ => Stream (Of (k, c)) m r+ -> Stream (Of (k, Seq.Seq c)) m r+groupByOrderedKey s = S.effect $ do+ (lkc S.:> r) <- S.toList s+ let hm = MS.fromListWith (<>) $ fmap (second Seq.singleton) lkc+ return $ MS.foldrWithKey (\k lc s' -> S.cons (k, lc) s') (return r) hm+{-# INLINABLE groupByOrderedKey #-}++-- | Wrap @Stream (Of d) m ()@ in a type which has @d@ as its last parameter+newtype StreamResult m d = StreamResult { unRes :: Stream (Of d) m () }++-- | get a @[]@ result from a Stream+resultToList :: Monad m => StreamResult m d -> m [d]+resultToList = S.toList_ . unRes++concatStreaming :: (Monad m, Monoid a) => Stream (Of a) m () -> m a+concatStreaming = S.iterT g . fmap (const mempty)+ where g (a S.:> ma) = fmap (a <>) ma++-- | @mappend@ all elements of a @StreamResult@ of monoids+concatStream :: (Monad m, Monoid a) => StreamResult m a -> m a+concatStream = concatStreaming . unRes++-- | apply monoidal stream-concatenation to a fold returning a stream to produce a fold returning the monoid+concatStreamFold+ :: Monoid b => FL.Fold a (StreamResult Identity b) -> FL.Fold a b+concatStreamFold = fmap (S.runIdentity . concatStream)++-- | apply monoidal stream-concatenation to an effectful fold returning a stream to produce an effectful fold returning the monoid+concatStreamFoldM+ :: (Monad m, Monoid b) => FL.FoldM m a (StreamResult m b) -> FL.FoldM m a b+concatStreamFoldM = MRC.postMapM concatStream++-- | map-reduce-fold builder returning a @StreamResult@+streamingEngine+ :: (Foldable g, Functor g)+ => ( forall z r+ . Stream (Of (k, z)) Identity r+ -> Stream (Of (k, g z)) Identity r+ )+ -> MRE.MapReduceFold y k c (StreamResult Identity) x d+streamingEngine groupByKey u (MRC.Assign a) r = fmap StreamResult $ FL.Fold+ (flip S.cons)+ (return ())+ ( S.map (\(k, lc) -> MRE.reduceFunction r k lc)+ . groupByKey+ . S.map a+ . unpackStream u+ )+{-# INLINABLE streamingEngine #-}++-- | effectful map-reduce-fold builder returning a @StreamResult@+streamingEngineM+ :: (Monad m, Traversable g)+ => (forall z r . Stream (Of (k, z)) m r -> Stream (Of (k, g z)) m r)+ -> MRE.MapReduceFoldM m y k c (StreamResult m) x d+streamingEngineM groupByKey u (MRC.AssignM a) r =+ fmap StreamResult . FL.generalize $ FL.Fold+ (flip S.cons)+ (return ())+ ( S.mapM (\(k, lc) -> MRE.reduceFunctionM r k lc)+ . groupByKey+ . S.mapM a+ . unpackStreamM u+ )+{-# INLINABLE streamingEngineM #-}+
+ src/Control/MapReduce/Engines/Streamly.hs view
@@ -0,0 +1,237 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE TypeApplications #-}+{-# LANGUAGE BangPatterns #-}+{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}+{-|+Module : Control.MapReduce.Engines.Streams+Description : map-reduce-folds builders+Copyright : (c) Adam Conner-Sax 2019+License : BSD-3-Clause+Maintainer : adam_conner_sax@yahoo.com+Stability : experimental++map-reduce engine (fold builder) using @Streamly@ streams as its intermediate and return type.++Notes:+1. These are polymorphic in the return stream type. Thought the streams do have to be @serial@ when @groupBy@ is called+So you have to specify the stream type in the call or it has to be inferrable from the use of the result.++2. There is a concurrent engine here, one that uses Streamly's concurrency features to map over the stream. I've not+been able to verify that this is faster on an appropriate task with appropriate runtime settings.+-}+module Control.MapReduce.Engines.Streamly+ (+ -- * Engines+ streamlyEngine+ , streamlyEngineM+ , concurrentStreamlyEngine++ -- * Result Extraction+ , resultToList+ , concatStream+ , concatStreamFold+ , concatStreamFoldM+ , concatConcurrentStreamFold++ -- * @groupBy@ Functions+ , groupByHashableKey+ , groupByOrderedKey+ , groupByHashableKeyST+ , groupByDiscriminatedKey++ -- * Re-Exports+ , SerialT+ , WSerialT+ , AheadT+ , AsyncT+ , WAsyncT+ , ParallelT+ , MonadAsync+ , IsStream+ )+where++import qualified Control.MapReduce.Core as MRC+import qualified Control.MapReduce.Engines as MRE++import Control.Arrow ( second )+import qualified Control.Foldl as FL+import Control.Monad.ST as ST+import qualified Data.Discrimination.Grouping as DG+import qualified Data.Foldable as F+import Data.Functor.Identity ( Identity(runIdentity) )+import Data.Hashable ( Hashable )+import qualified Data.HashMap.Strict as HMS+import qualified Data.HashTable.Class as HT+import qualified Data.HashTable.ST.Cuckoo as HTC+import qualified Data.Map.Strict as MS+import qualified Data.Sequence as Seq+import qualified Streamly.Prelude as S+import qualified Streamly as S+import Streamly ( SerialT+ , WSerialT+ , AheadT+ , AsyncT+ , WAsyncT+ , ParallelT+ , MonadAsync+ , IsStream+ )++-- | unpack for streamly based map/reduce+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)+{-# INLINABLE unpackStream #-}++-- | effectful (monadic) unpack for streamly based map/reduce+unpackStreamM :: (S.IsStream t, Monad m) => MRC.UnpackM m x y -> t m x -> t m y+unpackStreamM (MRC.FilterM t) = S.filterM t+unpackStreamM (MRC.UnpackM f) = S.concatMapM (fmap S.fromFoldable . f)+{-# INLINABLE unpackStreamM #-}++-- | make a stream into an (effectful) @[]@+resultToList :: (Monad m, S.IsStream t) => t m a -> m [a]+resultToList = S.toList . S.adapt++-- | mappend all in a monoidal stream+concatStream :: (Monad m, Monoid a) => S.SerialT m a -> m a+concatStream = S.foldl' (<>) mempty++-- | mappend everything in a pure Streamly fold+concatStreamFold :: Monoid b => FL.Fold a (S.SerialT Identity b) -> FL.Fold a b+concatStreamFold = fmap (runIdentity . concatStream)++-- | mappend everything in an effectful Streamly fold.+concatStreamFoldM+ :: (Monad m, Monoid b, S.IsStream t) => FL.FoldM m a (t m b) -> FL.FoldM m a b+concatStreamFoldM = MRC.postMapM (concatStream . S.adapt)++-- | mappend everything in a concurrent Streamly fold.+concatConcurrentStreamFold+ :: (Monad m, Monoid b, S.IsStream t) => FL.Fold a (t m b) -> FL.FoldM m a b+concatConcurrentStreamFold = concatStreamFoldM . FL.generalize++-- | map-reduce-fold builder returning a @SerialT Identity d@ result+streamlyEngine+ :: (Foldable g, Functor g)+ => (forall z . S.SerialT Identity (k, z) -> S.SerialT Identity (k, g z))+ -> MRE.MapReduceFold y k c (SerialT Identity) x d+streamlyEngine groupByKey u (MRC.Assign a) r = FL.Fold+ (flip S.cons)+ S.nil+ ( S.map (\(k, lc) -> MRE.reduceFunction r k lc)+ . groupByKey+ . S.map a+ . unpackStream u+ )+{-# INLINABLE streamlyEngine #-}++-- | unpack for concurrent streamly based map/reduce+unpackConcurrently+ :: (S.MonadAsync m, S.IsStream t) => MRC.Unpack x y -> t m x -> t m y+unpackConcurrently (MRC.Filter t) = S.filter t+unpackConcurrently (MRC.Unpack f) = S.concatMap (S.fromFoldable . f)+{-# INLINABLE unpackConcurrently #-}++-- | possibly (depending on chosen stream types) concurrent map-reduce-fold builder returning an @(Istream t, MonadAsync m) => t m d@ result+concurrentStreamlyEngine+ :: forall tIn tOut m g y k c x d+ . (S.IsStream tIn, S.IsStream tOut, S.MonadAsync m, Foldable g, Functor g)+ => (forall z . S.SerialT m (k, z) -> S.SerialT m (k, g z))+ -> MRE.MapReduceFold y k c (tOut m) x d+concurrentStreamlyEngine groupByKey u (MRC.Assign a) r = FL.Fold+ (\s a' -> (return a') `S.consM` s)+ S.nil+ ( S.mapM (\(k, lc) -> return $ MRE.reduceFunction r k lc)+ . S.adapt @SerialT @tOut -- make it concurrent for reducing+ . groupByKey+ . S.adapt @tIn @SerialT-- make it serial for grouping+ . S.mapM (return . a)+ . (S.|$) (unpackConcurrently 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+ :: (S.IsStream t, Monad m, S.MonadAsync m, Traversable g)+ => (forall z . SerialT m (k, z) -> SerialT m (k, g z))+ -> MRE.MapReduceFoldM m y k c (t m) x d+streamlyEngineM groupByKey u (MRC.AssignM a) r =+ FL.generalize+ $ fmap S.adapt+ $ FL.Fold+ (flip S.cons)+ S.nil+ ( S.mapM (\(k, lc) -> MRE.reduceFunctionM r k lc)+ . groupByKey -- this requires a serial stream.+ . S.mapM a+ . unpackStreamM u+ )+{-# INLINABLE streamlyEngineM #-}++-- TODO: Try using Streamly folds and Map.insertWith instead of toList and fromListWith. Prolly the same.+-- | Group streamly stream of @(k,c)@ by @hashable@ key.+-- NB: this function uses the fact that @SerialT m@ is a monad+groupByHashableKey+ :: (Monad m, Hashable k, Eq k)+ => S.SerialT m (k, c)+ -> S.SerialT m (k, Seq.Seq c)+groupByHashableKey s = do+ lkc <- S.yieldM (S.toList s)+ let hm = HMS.fromListWith (<>) $ fmap (second $ Seq.singleton) lkc+ HMS.foldrWithKey (\k lc s' -> S.cons (k, lc) s') S.nil hm+{-# INLINABLE groupByHashableKey #-}++-- TODO: Try using Streamly folds and Map.insertWith instead of toList and fromListWith. Prolly the same.+-- | Group streamly stream of @(k,c)@ by ordered key.+-- NB: this function uses the fact that @SerialT m@ is a monad+groupByOrderedKey+ :: (Monad m, Ord k) => S.SerialT m (k, c) -> S.SerialT m (k, Seq.Seq c)+groupByOrderedKey s = do+ lkc <- S.yieldM (S.toList s)+ let hm = MS.fromListWith (<>) $ fmap (second $ Seq.singleton) lkc+ MS.foldrWithKey (\k lc s' -> S.cons (k, lc) s') S.nil hm+{-# INLINABLE groupByOrderedKey #-}++-- | 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+ :: (Monad m, Hashable k, Eq k)+ => S.SerialT m (k, c)+ -> S.SerialT m (k, Seq.Seq c)+groupByHashableKeyST st = do+ lkc <- S.yieldM (S.toList st)+ ST.runST $ do+ hm <- (MRE.fromListWithHT @HTC.HashTable) (<>)+ $ fmap (second Seq.singleton) lkc+ HT.foldM (\s' (k, sc) -> return $ S.cons (k, sc) s') S.nil hm+{-# 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.SerialT m (k, c)+ -> S.SerialT m (k, Seq.Seq c)+groupByDiscriminatedKey s = do+ lkc <- S.yieldM (S.toList s)+ let g :: [(k, c)] -> (k, Seq.Seq c)+ g x = let k = fst (head x) in (k, F.fold $ fmap (Seq.singleton . snd) x)+ S.fromFoldable $ fmap g $ DG.groupWith fst lkc+{-# INLINABLE groupByDiscriminatedKey #-}
+ src/Control/MapReduce/Engines/Vector.hs view
@@ -0,0 +1,117 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE TypeApplications #-}+{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}+{-|+Module : Control.MapReduce.Engines.Vector+Description : map-reduce-folds builders+Copyright : (c) Adam Conner-Sax 2019+License : BSD-3-Clause+Maintainer : adam_conner_sax@yahoo.com+Stability : experimental++map-reduce engine (fold builder) using @Vector@ as its intermediate type.+-}+module Control.MapReduce.Engines.Vector+ (+ -- * Engines+ vectorEngine+ , vectorEngineM++ -- * groupBy functions+ , groupByHashableKey+ , groupByOrderedKey++ -- * re-exports+ , toList+ )+where++import qualified Control.MapReduce.Core as MRC+import qualified Control.MapReduce.Engines as MRE++import qualified Control.Foldl as FL+import Control.Monad ( (<=<) )+import qualified Data.Foldable as F+import Data.Hashable ( Hashable )+import qualified Data.HashMap.Strict as HMS+import qualified Data.Map.Strict as MS+import qualified Data.Sequence as Seq+import qualified Data.Vector as V+import Data.Vector ( Vector+ , toList+ )+import Control.Arrow ( second )++-- | case analysis of @Unpack@ for @Vector@ based mapReduce+unpackVector :: MRC.Unpack x y -> Vector x -> Vector y+unpackVector (MRC.Filter t) = V.filter t+unpackVector (MRC.Unpack f) = V.concatMap (V.fromList . F.toList . f)+{-# INLINABLE unpackVector #-}++-- | case analysis of @Unpack@ for @Vector@ based mapReduce+unpackVectorM :: Monad m => MRC.UnpackM m x y -> Vector x -> m (Vector y)+unpackVectorM (MRC.FilterM t) = V.filterM t+unpackVectorM (MRC.UnpackM f) =+ fmap (V.concatMap id) . traverse (fmap (V.fromList . F.toList) . f)+{-# INLINABLE unpackVectorM #-}+++-- | group the mapped and assigned values by key using a @Data.HashMap.Strict@+groupByHashableKey+ :: forall k c . (Hashable k, Eq k) => Vector (k, c) -> Vector (k, Seq.Seq c)+groupByHashableKey v =+ let hm = HMS.fromListWith (<>) $ V.toList $ fmap (second Seq.singleton) v+ in V.fromList $ HMS.toList hm -- HML.foldrWithKey (\k lc v -> V.snoc v (k,lc)) V.empty hm +{-# INLINABLE groupByHashableKey #-}++-- | group the mapped and assigned values by key using a @Data.Map.Strict@+groupByOrderedKey+ :: forall k c . Ord k => Vector (k, c) -> Vector (k, Seq.Seq c)+groupByOrderedKey v =+ let hm = MS.fromListWith (<>) $ V.toList $ fmap (second Seq.singleton) v+ in V.fromList $ MS.toList hm --MS.foldrWithKey (\k lc s -> VS.cons (k,lc) s) VS.empty hm+{-# INLINABLE groupByOrderedKey #-}++-- | map-reduce-fold builder, using @Vector@, returning a @Vector@ result+vectorEngine+ :: (Foldable g, Functor g)+ => (Vector (k, c) -> Vector (k, g c))+ -> MRE.MapReduceFold y k c Vector x d+vectorEngine groupByKey u (MRC.Assign a) r = fmap+ ( V.map (uncurry (MRE.reduceFunction r))+ . groupByKey+ . V.map a+ . unpackVector u+ )+ FL.vector+{-# INLINABLE vectorEngine #-}++-- | effectful map-reduce-fold builder, using @Vector@, returning an effectful @Vector@ result+vectorEngineM+ :: (Monad m, Traversable g)+ => (Vector (k, c) -> Vector (k, g c))+ -> MRE.MapReduceFoldM m y k c Vector x d+vectorEngineM groupByKey u (MRC.AssignM a) r = MRC.postMapM+ ( (traverse (uncurry (MRE.reduceFunctionM r)) =<<)+ . fmap groupByKey+ . (V.mapM a <=< unpackVectorM u)+ )+ (FL.generalize FL.vector)+{-# INLINABLE vectorEngineM #-}+-- NB: If we are willing to constrain to PrimMonad m, then we can use vectorM here which can do in-place updates, etc.+++
+ src/Control/MapReduce/Simple.hs view
@@ -0,0 +1,222 @@+{-# LANGUAGE DataKinds #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE GADTs #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE PolyKinds #-}+{-# LANGUAGE TypeFamilies #-}+{-# LANGUAGE ConstraintKinds #-}+{-# LANGUAGE MultiParamTypeClasses #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE UndecidableInstances #-}+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE InstanceSigs #-}+{-# LANGUAGE BangPatterns #-}+{-# OPTIONS_GHC -fwarn-incomplete-patterns #-}+{-|+Module : Control.MapReduce.Simple+Description : Simplified interfaces and helper functions for map-reduce-folds+Copyright : (c) Adam Conner-Sax 2019+License : BSD-3-Clause+Maintainer : adam_conner_sax@yahoo.com+Stability : experimental++Helper functions and default Engines and grouping functions for assembling map/reduce folds.+-}+module Control.MapReduce.Simple+ (+ -- * Unpackers+ noUnpack+ , simpleUnpack+ , filterUnpack++ -- * Assigners+ , assign++ -- * Reducers+ -- $reducers+ , processAndLabel+ , processAndLabelM+ , foldAndLabel+ , foldAndLabelM++ -- * Reduce Transformers + , reduceMapWithKey+ , reduceMMapWithKey++ -- * Default Map-Reduce Folds to @[]@+ , mapReduceFold+ , mapReduceFoldM+ , hashableMapReduceFold+ , hashableMapReduceFoldM+ , unpackOnlyFold+ , unpackOnlyFoldM++ -- * Simplify Results+ , concatFold+ , concatFoldM++ -- * Re-Exports+ , module Control.MapReduce.Core+ , Hashable+ )+where++import qualified Control.MapReduce.Core as MR+import Control.MapReduce.Core -- for re-export. I don't like the unqualified-ness of this.+import qualified Control.MapReduce.Engines.Streaming+ as MRST+import qualified Control.MapReduce.Engines.Streamly+ as MRSL+import qualified Control.MapReduce.Engines.List+ as MRL++--import qualified Control.MapReduce.Parallel as MRP++import qualified Control.Foldl as FL+import qualified Data.Foldable as F+import Data.Functor.Identity ( Identity(Identity)+ , runIdentity+ )++import Data.Hashable ( Hashable )++-- | Don't do anything in the unpacking stage+noUnpack :: MR.Unpack x x+noUnpack = MR.Filter $ const True+{-# INLINABLE noUnpack #-}++-- | unpack using the given function+simpleUnpack :: (x -> y) -> MR.Unpack x y+simpleUnpack f = MR.Unpack $ Identity . f+{-# INLINABLE simpleUnpack #-}++-- | Filter while unpacking, using the given function+filterUnpack :: (x -> Bool) -> MR.Unpack x x+filterUnpack = MR.Filter+{-# INLINABLE filterUnpack #-}++-- | Assign via two functions of @y@, one that provides the key and one that provides the data to be grouped by that key.+assign :: forall k y c . (y -> k) -> (y -> c) -> MR.Assign k y c+assign getKey getCols = let f !y = (getKey y, getCols y) in MR.Assign f+{-# INLINABLE assign #-}++-- | map a reduce using the given function of key and reduction result. +reduceMapWithKey :: (k -> y -> z) -> MR.Reduce k x y -> MR.Reduce k x z+reduceMapWithKey f r = case r of+ MR.Reduce g -> MR.Reduce $ \k -> fmap (f k) (g k)+ MR.ReduceFold gf -> MR.ReduceFold $ \k -> fmap (f k) (gf k)+{-# INLINABLE reduceMapWithKey #-}++-- | map a monadic reduction with a (non-monadic) function of the key and reduction result+reduceMMapWithKey :: (k -> y -> z) -> MR.ReduceM m k x y -> MR.ReduceM m k x z+reduceMMapWithKey f r = case r of+ MR.ReduceM g -> MR.ReduceM $ \k -> fmap (fmap (f k)) (g k)+ MR.ReduceFoldM gf -> MR.ReduceFoldM $ \k -> fmap (f k) (gf k)+{-# INLINABLE reduceMMapWithKey #-}+++{- $reducers+The most common case is that the reduction doesn't depend on the key.+These functions combine a key-independent processing step and a labeling step for the four variations of @Reduce@.+-}++-- | create a Reduce from a function of the grouped data to y and a function from the key and y to the result type+processAndLabel+ :: (forall h . (Foldable h, Functor h) => h x -> y)+ -> (k -> y -> z)+ -> MR.Reduce k x z+processAndLabel process relabel = MR.Reduce $ \k -> relabel k . process+{-# INLINABLE processAndLabel #-}++-- | create a monadic ReduceM from a function of the grouped data to (m y) and a function from the key and y to the result type+processAndLabelM+ :: Monad m+ => (forall h . (Foldable h, Functor h) => h x -> m y)+ -> (k -> y -> z)+ -> MR.ReduceM m k x z+processAndLabelM processM relabel =+ MR.ReduceM $ \k -> fmap (relabel k) . processM+{-# INLINABLE processAndLabelM #-}++-- | create a Reduce from a fold of the grouped data to y and a function from the key and y to the result type+foldAndLabel :: FL.Fold x y -> (k -> y -> z) -> MR.Reduce k x z+foldAndLabel fld relabel = let q !k = fmap (relabel k) fld in MR.ReduceFold q+{-# INLINABLE foldAndLabel #-}++-- | create a monadic ReduceM from a monadic fold of the grouped data to (m y) and a function from the key and y to the result type+foldAndLabelM+ :: Monad m => FL.FoldM m x y -> (k -> y -> z) -> MR.ReduceM m k x z+foldAndLabelM fld relabel =+ let q !k = fmap (relabel k) fld in MR.ReduceFoldM q+{-# INLINABLE foldAndLabelM #-}++-- | The simple fold types return lists of results. Often we want to merge these into some other structure via (<>)+concatFold :: (Monoid d, Foldable g) => FL.Fold a (g d) -> FL.Fold a d+concatFold = fmap F.fold++-- | The simple fold types return lists of results. Often we want to merge these into some other structure via (<>)+concatFoldM+ :: (Monad m, Monoid d, Foldable g) => FL.FoldM m a (g d) -> FL.FoldM m a d+concatFoldM = fmap F.fold++mapReduceFold+ :: Ord k+ => MR.Unpack x y -- ^ unpack x to none or one or many y's+ -> MR.Assign k y c -- ^ assign each y to a key value pair (k,c)+ -> MR.Reduce k c d -- ^ reduce a grouped [c] to d+ -> FL.Fold x [d]+mapReduceFold u a r =+ fmap (runIdentity . MRSL.resultToList)+ $ MRSL.streamlyEngine MRSL.groupByOrderedKey u a r+{-# INLINABLE mapReduceFold #-}++mapReduceFoldM+ :: (Monad m, Ord k)+ => MR.UnpackM m x y -- ^ unpack x to none or one or many y's+ -> MR.AssignM m k y c -- ^ assign each y to a key value pair (k,c)+ -> MR.ReduceM m k c d -- ^ reduce a grouped [c] to d+ -> FL.FoldM m x [d]+mapReduceFoldM u a r =+ MR.postMapM id $ fmap MRST.resultToList $ MRST.streamingEngineM+ MRST.groupByOrderedKey+ u+ a+ r+{-# INLINABLE mapReduceFoldM #-}++hashableMapReduceFold+ :: (Hashable k, Eq k)+ => MR.Unpack x y -- ^ unpack x to none or one or many y's+ -> MR.Assign k y c -- ^ assign each y to a key value pair (k,c)+ -> MR.Reduce k c d -- ^ reduce a grouped [c] to d+ -> FL.Fold x [d]+hashableMapReduceFold u a r =+ fmap (runIdentity . MRSL.resultToList)+ $ MRSL.streamlyEngine MRSL.groupByHashableKey u a r+{-# INLINABLE hashableMapReduceFold #-}++hashableMapReduceFoldM+ :: (Monad m, Hashable k, Eq k)+ => MR.UnpackM m x y -- ^ unpack x to to none or one or many y's+ -> MR.AssignM m k y c -- ^ assign each y to a key value pair (k,c)+ -> MR.ReduceM m k c d -- ^ reduce a grouped [c] to d+ -> FL.FoldM m x [d]+hashableMapReduceFoldM u a r =+ MR.postMapM id $ fmap MRST.resultToList $ MRST.streamingEngineM+ MRST.groupByHashableKey+ u+ a+ r+{-# INLINABLE hashableMapReduceFoldM #-}++-- | do only the unpack step.+unpackOnlyFold :: MR.Unpack x y -> FL.Fold x [y]+unpackOnlyFold u = fmap (MRL.unpackList u) FL.list+{-# INLINABLE unpackOnlyFold #-}++-- | do only the (monadic) unpack step. Use a TypeApplication to specify what to unpack to. As in 'unpackOnlyFoldM @[]'+unpackOnlyFoldM :: Monad m => MR.UnpackM m x y -> FL.FoldM m x [y]+unpackOnlyFoldM u = MR.postMapM (MRL.unpackListM u) (FL.generalize FL.list)+{-# INLINABLE unpackOnlyFoldM #-}
+ test/Test1.hs view
@@ -0,0 +1,66 @@+{-# LANGUAGE OverloadedStrings #-}+module Test1+ ( tests+ )+where++import Hedgehog+import qualified Hedgehog.Gen as Gen+import qualified Hedgehog.Range as Range++import qualified Data.Map as M+import qualified Data.List as L+import qualified Data.Foldable as F+import qualified Control.Foldl as FL++import qualified Control.MapReduce.Simple as MR+import qualified Control.MapReduce.Engines.List+ as MRL+import qualified Control.MapReduce.Engines.Vector+ as MRV+import qualified Control.MapReduce.Engines.Streamly+ as MRSY+import qualified Control.MapReduce.Engines.Streaming+ as MRSG++-- for reference+direct :: [Int] -> M.Map Bool Int+direct = M.fromListWith (+) . fmap (\x -> (x `mod` 3 == 0, x)) . L.filter even++unpack = MR.filterUnpack even -- filter, keeping even numbers++-- assign each number a key of True or False depending on whether it's a multiple of 3. Group the number itself.+isMultipleOf3 x = (x `mod` 3) == 0+assign = MR.assign isMultipleOf3 id++reduce = MR.foldAndLabel FL.sum M.singleton -- sum each group and then create a singleton Map for later combining++mrFoldList :: FL.Fold Int (M.Map Bool Int)+mrFoldList =+ fmap F.fold $ MRL.listEngine MRL.groupByHashableKey unpack assign reduce++mrFoldVector :: FL.Fold Int (M.Map Bool Int)+mrFoldVector =+ fmap F.fold $ MRV.vectorEngine MRV.groupByHashableKey unpack assign reduce++mrFoldStreamly :: FL.Fold Int (M.Map Bool Int)+mrFoldStreamly = MRSY.concatStreamFold+ $ MRSY.streamlyEngine MRSY.groupByHashableKey unpack assign reduce++mrFoldStreaming :: FL.Fold Int (M.Map Bool Int)+mrFoldStreaming = MRSG.concatStreamFold+ $ MRSG.streamingEngine MRSG.groupByHashableKey unpack assign reduce++prop_mrSame :: FL.Fold Int (M.Map Bool Int) -> Property+prop_mrSame mrF = property $ do+ xs <- forAll $ Gen.list (Range.linear 0 100) $ Gen.int (Range.linear 0 10000)+ direct xs === FL.fold mrF xs++tests :: Group+tests = Group+ "Test 1"+ [ ("[] Engine" , prop_mrSame mrFoldList)+ , ("Vector Engine" , prop_mrSame mrFoldVector)+ , ("Streamly Engine" , prop_mrSame mrFoldStreamly)+ , ("Stremaing Engine", prop_mrSame mrFoldStreaming)+ ]
+ test/TestAll.hs view
@@ -0,0 +1,10 @@++{-# LANGUAGE OverloadedStrings #-}+module Main where++import Hedgehog++import Test1 ( tests )++main :: IO Bool+main = checkParallel tests