Holumbus-MapReduce-0.1.1: source/Holumbus/MapReduce/Types.hs
-- ----------------------------------------------------------------------------
{- |
Module : Holumbus.MapReduce.Types
Copyright : Copyright (C) 2008 Stefan Schmidt
License : MIT
Maintainer : Stefan Schmidt (stefanschmidt@web.de)
Stability : experimental
Portability: portable
Version : 0.1
-}
-- ----------------------------------------------------------------------------
{-# OPTIONS -fglasgow-exts #-}
{-# LANGUAGE Arrows, NoMonomorphismRestriction #-}
module Holumbus.MapReduce.Types
(
Hash(..)
, hashedPartition
, FunctionData(..)
{-
, encodeTuple
, decodeTuple
, decodeTupleList
, encodeTupleList
-}
-- * TaskData
, TaskId
, TaskType(..)
, TaskState(..)
, getNextTaskState
, TaskOutputType(..)
, TaskData(..)
-- * JobData
, JobId
, JobState(..)
, getNextJobState
, getPrevJobState
, fromJobStatetoTaskType
, OutputMap
, JobAction(..)
, JobInfo(..)
, JobData(..)
, JobResultContainer(..)
, JobResult(..)
-- * TaskAction
, ActionName
, ActionInfo
, ActionEnvironment(..)
, mkActionEnvironment
, InputReader
, OutputWriter
, OptionsDecoder
-- ----------------------------------------------------------------------------
-- remove this, when work done!
-- defaultFunctions (only for profiling)
, defaultInputReader
, defaultOutputWriter
, defaultSplit
, defaultPartition
, defaultMerge
, readConnector
, writeConnector
-- ----------------------------------------------------------------------------
, ActionConfiguration(..)
, SplitConfiguration(..)
, MapConfiguration(..)
, ReduceConfiguration(..)
, defaultActionConfiguration
, defaultSplitConfiguration
, defaultMapConfiguration
, defaultReduceConfiguration
, readActionConfiguration
, createJobInfoFromConfiguration
, createListsFromJobResult
, ActionData(..)
, getActionForTaskType
, ActionMap
-- * MapAction
, MapAction
-- , BinaryMapAction
, MapFunction
, MapPartition
-- * Combine-Reduce-Action
, ReduceAction
-- , BinaryReduceAction
, ReduceMerge
, ReduceFunction
, ReducePartition
, SplitFunction
, SplitAction
)
where
import Control.Concurrent
import Data.Binary
--import Holumbus.Common.MRBinary
import qualified Data.ByteString.Lazy as B
import qualified Data.Map as Map
import qualified Data.HashTable as Hash
import Data.Maybe
import Data.Time
import System.Log.Logger
import Data.Time.Clock.POSIX
import Text.XML.HXT.Arrow
import qualified Holumbus.Data.AccuMap as AMap
import Holumbus.Common.FileHandling
import Holumbus.Common.Utils
import qualified Holumbus.Data.KeyMap as KMap
import qualified Holumbus.FileSystem.FileSystem as FS
import Control.DeepSeq
localLogger :: String
localLogger = "Holumbus.MapReduce.Types"
-- ----------------------------------------------------------------------------
-- general datatypes
-- ----------------------------------------------------------------------------
class Hash a where
hash :: Int -> a -> Int
instance Hash Int where
hash n k = mod k n
data FunctionData
= TupleFunctionData B.ByteString
| FileFunctionData FS.FileId
deriving (Show, Eq, Ord)
instance Binary FunctionData where
put (TupleFunctionData t) = putWord8 0 >> put t
put (FileFunctionData f) = putWord8 1 >> put f
get
= do
t <- getWord8
case t of
1 -> get >>= \f -> return (FileFunctionData f)
_ -> get >>= \b -> return (TupleFunctionData b)
instance XmlPickler FunctionData where
xpickle = xpFunctionData
xpFunctionData :: PU FunctionData
xpFunctionData
= xpElem "data" $
xpAlt tag ps
where
tag (TupleFunctionData _) = 0
tag (FileFunctionData _) = 1
ps = [xpWrap (\p -> TupleFunctionData (encodeTuple p), \(TupleFunctionData p) -> decodeTuple p) xpRawFunctionData
,xpWrap (\f -> FileFunctionData f, \(FileFunctionData f) -> f) xpFileFunctionData ]
xpRawFunctionData = xpPair (xpElem "key" xpText0) (xpElem "value" xpText0)
xpFileFunctionData = xpElem "filename" xpText
-- ----------------------------------------------------------------------------
-- general encoding / decoding
-- ----------------------------------------------------------------------------
encodeTuple :: (Binary k, Binary v) => (k, v) -> B.ByteString
encodeTuple t = encode t
decodeTuple :: (Binary k, Binary v) => B.ByteString -> (k, v)
decodeTuple t = decode t
-- encodeTupleList :: (Binary k, Binary v) => [(k, v)] -> [B.ByteString]
-- encodeTupleList ls = map encodeTuple ls
-- decodeTupleList :: (Binary k, Binary v) => [B.ByteString] -> [(k, v)]
-- decodeTupleList ls = map decodeTuple ls
-- ----------------------------------------------------------------------------
-- Task DataTypes
-- ----------------------------------------------------------------------------
-- | the task id (should be unique in the system)
type TaskId = Integer
-- | which type (map, combine, reduce)
data TaskType = TTSplit | TTMap | TTCombine | TTReduce | TTError
deriving (Show, Eq, Ord)
instance Binary TaskType where
put (TTSplit) = putWord8 1
put (TTMap) = putWord8 2
put (TTCombine) = putWord8 3
put (TTReduce) = putWord8 4
put (TTError) = putWord8 0
get
= do
t <- getWord8
case t of
1 -> return (TTSplit)
2 -> return (TTMap)
3 -> return (TTCombine)
4 -> return (TTReduce)
_ -> return (TTError)
-- | the task state
data TaskState = TSIdle | TSSending | TSInProgress | TSCompleted | TSFinished | TSError
deriving (Show, Eq, Ord, Enum)
instance Binary TaskState where
put (TSIdle) = putWord8 1
put (TSSending) = putWord8 2
put (TSInProgress) = putWord8 3
put (TSCompleted) = putWord8 4
put (TSFinished) = putWord8 5
put (TSError) = putWord8 0
get
= do
t <- getWord8
case t of
1 -> return (TSIdle)
2 -> return (TSSending)
3 -> return (TSInProgress)
4 -> return (TSCompleted)
5 -> return (TSFinished)
_ -> return (TSError)
getNextTaskState :: TaskState -> TaskState
getNextTaskState TSError = TSError
getNextTaskState TSFinished = TSFinished
getNextTaskState s = succ s
data TaskOutputType = TOTRawTuple | TOTFile
deriving (Show, Read, Eq, Ord, Enum)
instance Binary TaskOutputType where
put (TOTRawTuple) = putWord8 0
put (TOTFile) = putWord8 1
get
= do
t <- getWord8
case t of
0 -> return (TOTRawTuple)
_ -> return (TOTFile)
instance XmlPickler TaskOutputType where
xpickle = xpAttr "output" $ xpPrim
-- | the TaskData, contains all information to do the task
data TaskData = TaskData {
td_JobId :: ! JobId
, td_TaskId :: ! TaskId
, td_Type :: ! TaskType
, td_State :: ! TaskState
, td_Option :: ! B.ByteString
, td_PartValue :: ! (Maybe Int)
, td_Input :: ! (Int, [FunctionData])
, td_Output :: ! [(Int,[FunctionData])]
, td_OutputType :: ! TaskOutputType
, td_Action :: ! ActionName
} deriving (Show, Eq, Ord)
instance Binary TaskData where
put (TaskData jid tid tt ts opt pv i o ot a)
= put jid >> put tid >> put tt >> put ts >> put opt >> put pv >> put i >> put o >> put ot >> put a
get
= do
jid <- get
tid <- get
tt <- get
ts <- get
opt <- get
pv <- get
i <- get
o <- get
ot <- get
a <- get
return (TaskData jid tid tt ts opt pv i o ot a)
-- ----------------------------------------------------------------------------
-- Job Datatypes
-- ----------------------------------------------------------------------------
-- | the job id (should be unique in the system)
type JobId = Integer
-- | the job state
data JobState = JSPlanned | JSIdle | JSSplit | JSMap | JSCombine | JSReduce | JSCompleted | JSFinished | JSError
deriving(Show, Eq, Ord, Enum)
instance Binary JobState where
put (JSPlanned) = putWord8 1
put (JSIdle) = putWord8 2
put (JSSplit) = putWord8 3
put (JSMap) = putWord8 4
put (JSCombine) = putWord8 5
put (JSReduce) = putWord8 6
put (JSCompleted) = putWord8 7
put (JSFinished) = putWord8 8
put (JSError) = putWord8 0
get
= do
t <- getWord8
case t of
1 -> return (JSPlanned)
2 -> return (JSIdle)
3 -> return (JSSplit)
4 -> return (JSMap)
5 -> return (JSCombine)
6 -> return (JSReduce)
7 -> return (JSCompleted)
8 -> return (JSFinished)
_ -> return (JSError)
getNextJobState :: JobState -> JobState
getNextJobState JSError = JSError
getNextJobState JSFinished = JSFinished
getNextJobState s = succ s
getPrevJobState :: JobState -> JobState
getPrevJobState JSIdle = JSIdle
getPrevJobState JSError = JSError
getPrevJobState s = pred s
fromJobStatetoTaskType :: JobState -> Maybe TaskType
fromJobStatetoTaskType JSSplit = Just TTSplit
fromJobStatetoTaskType JSMap = Just TTMap
fromJobStatetoTaskType JSCombine = Just TTCombine
fromJobStatetoTaskType JSReduce = Just TTReduce
fromJobStatetoTaskType _ = Nothing
type OutputMap = Map.Map JobState (AMap.AccuMap Int FunctionData)
data JobAction = JobAction {
ja_Name :: ! ActionName
, ja_Output :: ! TaskOutputType
, ja_Count :: ! Int
} deriving (Show, Eq)
instance Binary JobAction where
put (JobAction n o c) = put n >> put o >> put c
get
= do
n <- get
o <- get
c <- get
return (JobAction n o c)
instance XmlPickler JobAction where
xpickle = xpJobAction
xpJobAction :: PU JobAction
xpJobAction =
xpWrap
(\(n, o, c) -> JobAction n o c,
\(JobAction n o c) -> (n, o, c)) xpAction
where
xpAction = xpTriple (xpFunction) (xpickle) (xpCount)
xpFunction = xpAttr "name" xpText
xpCount = xpWrap ((maybe 1 id), Just) $ xpOption $ xpAttr "count" xpickle
-- | defines a job, this is all data the user has to give to run a job
data JobInfo = JobInfo {
ji_Description :: ! String
, ji_Option :: ! B.ByteString
, ji_SplitAction :: ! (Maybe JobAction)
, ji_MapAction :: ! (Maybe JobAction)
, ji_CombineAction :: ! (Maybe JobAction)
, ji_ReduceAction :: ! (Maybe JobAction)
, ji_NumOfResults :: ! (Maybe Int)
, ji_Input :: ! [FunctionData]
} deriving (Show, Eq)
instance Binary JobInfo where
put (JobInfo d opt sa ma ca ra nor i)
= put d >> put opt >>
put sa >> put ma >> put ca >> put ra >> put nor >> put i
get
= do
d <- get
opt <- get
sa <- get
ma <- get
ca <- get
ra <- get
nor <- get
i <- get
return (JobInfo d opt sa ma ca ra nor i)
instance XmlPickler JobInfo where
xpickle = xpJobInfo
xpJobInfo :: PU JobInfo
xpJobInfo =
xpElem "jobInfo" $
xpWrap
(\(n, (sa,ma,ca,ra), nor, i) -> JobInfo n (encode ()) sa ma ca ra nor i,
\(JobInfo n _ sa ma ca ra nor i) -> (n, (sa,ma,ca,ra), nor, i)) xpJob
where
xpJob =
xp4Tuple
(xpAttr "name" xpText0)
(xpActions)
(xpCount)
(xpFunctionDataList)
xpActions = xp4Tuple (xpSplitAction) (xpMapAction) (xpCombineAction) (xpReduceAction)
xpSplitAction = xpOption $ xpElem "split" $ xpJobAction
xpMapAction = xpOption $ xpElem "map" $ xpJobAction
xpCombineAction = xpOption $ xpElem "combine" $ xpJobAction
xpReduceAction = xpOption $ xpElem "reduce" $ xpJobAction
xpFunctionDataList
= xpWrap (filterEmptyList,setEmptyList) $ xpOption $
xpElem "inputList" $ xpList xpFunctionData
xpCount = xpOption $ xpAttr "numOfResults" xpickle
-- | the job data, include the user-input and some additional control-data
data JobData = JobData {
jd_JobId :: JobId
, jd_State :: JobState
, jd_OutputMap :: OutputMap
, jd_Info :: JobInfo
, jd_startTime :: UTCTime
, jd_endTime :: UTCTime
, jd_Result :: JobResultContainer
}
instance Show JobData where
show (JobData jid state om _ t1 t2 _) =
"JobId:\t" ++ show jid ++ "\n" ++
"State:\t" ++ show state ++ "\n" ++
"Info:\tJobInfo\n" ++
"StartTime:\t" ++ show t1 ++ "\n" ++
"EndTime:\t" ++ show t2 ++ "\n" ++
"OutputMap:\n" ++ show om ++ "\n"
data JobResultContainer = JobResultContainer (MVar JobResult)
instance Show JobResultContainer where
show _ = "JobResultContainer"
-- | the result of the job, given by the master
data JobResult = JobResult {
jr_Output :: [FunctionData]
} deriving (Show)
instance Binary JobResult where
put (JobResult o) = put o
get
= do
o <- get
return (JobResult o)
-- ----------------------------------------------------------------------------
-- Reader and Writer
-- ----------------------------------------------------------------------------
-- | the ActionEnvironment contains all data that might be needed
-- during an action. So far, it only keeps the current task data and
-- a reference to the global filesystem and the options.
-- This is a good place to implement counters for the map-reduce-system
-- or other stuff...
data ActionEnvironment = ActionEnvironment {
ae_TaskData :: TaskData
, ae_FileSystem :: FS.FileSystem
}
mkActionEnvironment :: TaskData -> FS.FileSystem -> ActionEnvironment
mkActionEnvironment td fs = ActionEnvironment td fs
type InputReader k1 v1 = B.ByteString -> IO [(k1,v1)]
type OutputWriter k2 v2 = [(k2,v2)] -> IO B.ByteString
readConnector
:: (NFData k1, NFData v1, Binary k1, Binary v1)
=> InputReader k1 v1
-> ActionEnvironment
-> [FunctionData]
-> IO [(k1,v1)]
readConnector ic ae ls
= do
let (tuples,filenames) = splitList ls
mayBeTupleContent <- mapM readInput tuples
fileContent <- read2 filenames
return . concat $ (catMaybes mayBeTupleContent) ++ fileContent
where
-- split list into tuople data and filenames
splitList :: [FunctionData] -> ([B.ByteString],[FS.FileId])
splitList [] = ([],[])
splitList ((TupleFunctionData t):xs) = ((t:ts),fs)
where
(ts,fs) = splitList xs
splitList ((FileFunctionData f):xs) = (ts,f:fs)
where
(ts,fs) = splitList xs
-- read all files at once
--read2 :: [FS.FileId] -> IO [[(k1,v1)]]
read2 files = do
contents <- FS.getMultiFileContent files fs
encodedcontents <- mapM (ic . snd) contents
return encodedcontents
where
fs = ae_FileSystem ae
-- readInput :: FS.FileSystem -> FunctionData -> IO (Maybe [(k1,v1)])
readInput t = return $ Just $ [decode t]
{- readInput fs (FileFunctionData f)
= do
debugM localLogger $ "loadInputList: getting content for: " ++ f
mbc <- FS.getFileContent f fs
if isNothing mbc
then do
debugM localLogger $ "loadInputList: no content found"
return Nothing
else do
d <- ic $ fromJust mbc
return $ Just d
-}
--
--myMapM :: (a -> IO b) -> [a] -> IO [b]
--myMapM f [] = return []
--myMapM f (x:xs) = do
-- infoM localLogger $ "myMapM: do f"
-- b <- f x
-- infoM localLogger $ "myMapM: do recursiv map"
-- bs <- myMapM f xs
-- infoM localLogger $ "myMapM: do return"
-- return (b:bs)
writeConnector
:: (Binary k2, Binary v2)
=> OutputWriter k2 v2
-> ActionEnvironment
-> [(Int,[(k2,v2)])]
-> IO [(Int,[FunctionData])]
writeConnector oc ae ls
= do
infoM localLogger $ "writeConnector: " ++ (show . length $ ls)
os <- case tot of
TOTRawTuple -> mapM (writeOutput (ae_FileSystem ae) tot) ls
_ -> write2
return $ catMaybes os
where
write2 = do
bincontents <- mapM (\(_,c) -> oc c) ls
let filelist = zip filenames bincontents;
FS.createFiles filelist fs
return $ (zipWith (\(i,_) fn -> Just (i,[FileFunctionData fn]) ) ls filenames)
where
filenames = map (\(i,_) -> "j" ++ show (td_JobId td) ++ "_t" ++ show (td_TaskId td) ++ "_i" ++ show i) ls
fs = ae_FileSystem ae
td = ae_TaskData ae
tot = td_OutputType $ ae_TaskData ae
-- writeOutput :: FS.FileSystem -> TaskOutputType -> (Int,[(k2,v2)]) -> IO (Maybe (Int,[FunctionData]))
writeOutput _ TOTRawTuple (i,ts)
= return $ Just $ (i,bs)
where
bs = map (\t -> TupleFunctionData $ encode t) ts
-- TODO exception werfen
writeOutput fs _ (i,ts)
= do
infoM localLogger "oc ts"
c <- oc ts
infoM localLogger "appendfile"
FS.appendFile fn c fs
infoM localLogger "return just"
return $ Just (i,[FileFunctionData fn])
where
fn = "j" ++ show (td_JobId td) ++ "_t" ++ show (td_TaskId td) ++ "_i" ++ show i
defaultInputReader :: (NFData v1, NFData k1, Binary k1, Binary v1) => InputReader k1 v1
defaultInputReader = return . parseByteStringToList
defaultOutputWriter :: (NFData v2, NFData k2, Binary k2, Binary v2) => OutputWriter k2 v2
defaultOutputWriter = return . listToByteString
-- = return $ B.concat $ map encode ls
type ActionName = String
type ActionInfo = String
data ActionConfiguration a k1 v1 k2 v2 v3 v4
= ActionConfiguration {
ac_Name :: ActionName
, ac_Info :: ActionInfo
, ac_OptEncoder :: OptionsEncoder a
, ac_OptDecoder :: OptionsDecoder a
, ac_InputEncoder :: InputEncoder k1 v1
, ac_OutputDecoder :: OutputDecoder k2 v4
, ac_Split :: Maybe (SplitConfiguration a k1 v1)
, ac_Map :: Maybe (MapConfiguration a k1 v1 k2 v2)
, ac_Combine :: Maybe (ReduceConfiguration a k2 v2 v3)
, ac_Reduce :: Maybe (ReduceConfiguration a k2 v3 v4)
}
data SplitConfiguration a k1 v1
= SplitConfiguration {
sc_Function :: SplitFunction a k1 v1
, sc_Reader :: InputReader k1 v1
, sc_Writer :: OutputWriter k1 v1
}
data MapConfiguration a k1 v1 k2 v2
= MapConfiguration {
mc_Function :: MapFunction a k1 v1 k2 v2
, mc_Partition :: MapPartition a k2 v2
, mc_Reader :: InputReader k1 v1
, mc_Writer :: OutputWriter k2 v2
}
data ReduceConfiguration a k2 v3 v4
= ReduceConfiguration {
rc_Merge :: ReduceMerge a k2 v3
, rc_Function :: ReduceFunction a k2 v3 v4
, rc_Partition :: ReducePartition a k2 v4
, rc_Reader :: InputReader k2 v3
, rc_Writer :: OutputWriter k2 v4
}
defaultActionConfiguration
:: (NFData v1, NFData k1, Binary a, Binary k1, Binary v1, Binary k2, Binary v4)
=> ActionName
-> ActionConfiguration a k1 v1 k2 v2 v3 v4
defaultActionConfiguration name
= ActionConfiguration
name
""
defaultOptionsEncoder
defaultOptionsDecoder
defaultInputEncoder
defaultOutputDecoder
(Just defaultSplitConfiguration)
Nothing
Nothing
Nothing
defaultSplitConfiguration
:: (NFData v1, NFData k1, Binary a, Binary k1, Binary v1)
=> SplitConfiguration a k1 v1
defaultSplitConfiguration
= SplitConfiguration
defaultSplit
defaultInputReader
defaultOutputWriter
defaultMapConfiguration
:: (NFData v1, NFData k1, NFData v2, NFData k2, Ord k2, Binary a, Binary k1, Binary v1, Binary k2, Binary v2)
=> MapFunction a k1 v1 k2 v2
-> MapConfiguration a k1 v1 k2 v2
defaultMapConfiguration fct
= MapConfiguration
fct
defaultPartition
defaultInputReader
defaultOutputWriter
defaultReduceConfiguration
:: (NFData v2, NFData k2, NFData v3, Ord k2, Binary a, Binary k2, Binary v2, Binary v3)
=> ReduceFunction a k2 v2 v3
-> ReduceConfiguration a k2 v2 v3
defaultReduceConfiguration fct
= ReduceConfiguration
defaultMerge
fct
defaultPartition
defaultInputReader
defaultOutputWriter
defaultSplit
:: (NFData k1, NFData v1) => SplitFunction a k1 v1
defaultSplit _ _ n ls = do
infoM localLogger "defaultSplit"
return partedList
where
partedList = AMap.toList $ AMap.fromList ps
ns = [(x `mod` n) + 1 | x <- [0..]]
is = map (\a -> [a]) ls
ps = zip ns is
hashedPartition :: (Hash k2, Binary k2, Binary v2, NFData k2, NFData v2) => MapPartition a k2 v2
hashedPartition _ _ 1 l = return . (:[]) . (,) 1 $ l
hashedPartition _ _ n l = do
infoM localLogger "hashedPartition: map"
let a = map (\t -> (hash n (fst t),[t])) l
infoM localLogger "hashedPartition: fromList"
let b = AMap.fromList a
infoM localLogger "hashedPartition: toList"
let c = AMap.toList b
infoM localLogger "hashedPartition: return"
return $ c
defaultPartition
:: (Binary k2, Binary v2)
=> MapPartition a k2 v2
defaultPartition _ _ 1 ls
-- To make it faster, wenn no partition is used
= return [(1,ls)]
defaultPartition _ _ n ls
= do
-- calculate partition-Values
let markedList = map (\t@(k,_) -> (hash' k,t)) ls
-- merge them
-- TODO this might change (revert) the order of the Elements...
let resultList = AMap.toList $ AMap.fromTupleList markedList
return resultList
where
-- calculate a hash-value, because we only have the Binary-Instance, we
-- can only use the Bytestring of the Value
hash' k = ((fromIntegral $ Hash.hashString (show $ encode k)) `mod` n) + 1
defaultMerge
:: (Ord k2, Binary k2, Binary v2)
=> ReduceMerge a k2 v2
defaultMerge _ _ ls
= return $ AMap.toList $ AMap.fromTupleList ls
data ActionData
= ActionData {
ad_Name :: ActionName
, ad_Info :: ActionInfo
, ad_Split :: Maybe BinarySplitAction
, ad_Map :: Maybe BinaryMapAction
, ad_Combine :: Maybe BinaryReduceAction
, ad_Reduce :: Maybe BinaryReduceAction
}
instance KMap.Key ActionData where
getKey = ad_Name
instance Show ActionData where
show (ActionData n i _ _ _ _) = "{ActionData name:\"" ++ n ++ "\" info:\"" ++ i ++ "\"}"
type ActionMap = KMap.KeyMap ActionData
type OptionsEncoder a = a -> B.ByteString
type OptionsDecoder a = B.ByteString -> a
defaultOptionsEncoder :: (Binary a) => OptionsEncoder a
defaultOptionsEncoder = encode
defaultOptionsDecoder :: (Binary a) => OptionsDecoder a
defaultOptionsDecoder = decode
type InputEncoder k1 v1 = [(k1,v1)] -> [FS.FileId] -> [FunctionData]
type OutputDecoder k2 v4 = [FunctionData] -> ([(k2,v4)],[FS.FileId])
defaultInputEncoder :: (Binary k1, Binary v1) => InputEncoder k1 v1
defaultInputEncoder ls1 ls2 = ls1' ++ ls2'
where
ls1' = map (\t -> TupleFunctionData (encode t)) ls1
ls2' = map (\f -> FileFunctionData f) ls2
defaultOutputDecoder :: (Binary k2, Binary v4) => OutputDecoder k2 v4
defaultOutputDecoder ls = (ls1, ls2)
where
ls1 = map (\t -> decode t) $ catMaybes ls1'
ls2 = catMaybes ls2'
(ls1',ls2') = unzip $ map (splitter) ls
splitter (TupleFunctionData t) = (Just t, Nothing)
splitter (FileFunctionData f) = (Nothing, Just f)
getActionForTaskType :: TaskType -> ActionData -> Maybe BinaryReduceAction
getActionForTaskType TTSplit ad = ad_Split ad
getActionForTaskType TTMap ad = ad_Map ad
getActionForTaskType TTCombine ad = ad_Combine ad
getActionForTaskType TTReduce ad = ad_Reduce ad
getActionForTaskType _ _ = Nothing
readActionConfiguration
:: ( Ord k2, Binary a
, Show k1, Show v1
, Show k2, Show v2, Show v3, Show v4
, NFData k1, NFData v1
, NFData k2, NFData v2, NFData v3
, Binary k1, Binary v1
, Binary k2, Binary v2
, Binary v3, Binary v4)
=> ActionConfiguration a k1 v1 k2 v2 v3 v4
-> ActionData
readActionConfiguration (ActionConfiguration n i _ optDec _ _ sc mc cc rc)
= ActionData n i sf mf cf rf
where
sf = maybe Nothing (\(SplitConfiguration f ir ow) -> Just $ performSplitAction optDec f ir ow) sc
mf = maybe Nothing (\(MapConfiguration f p ir ow) -> Just $ performMapAction optDec f p ir ow) mc
cf = maybe Nothing (\(ReduceConfiguration m f p ir ow) -> Just $ performReduceAction optDec m f p ir ow) cc
rf = maybe Nothing (\(ReduceConfiguration m f p ir ow) -> Just $ performReduceAction optDec m f p ir ow) rc
createJobInfoFromConfiguration
:: ActionConfiguration a k1 v1 k2 v2 v3 v4
-> a -- ^ options
-> [(k1,v1)] -- ^ input (Tuples)
-> [FS.FileId] -- ^ input (Files)
-> Int -- ^ number of splitters
-> Int -- ^ number of mappers
-> Int -- ^ number of reducers
-> Int -- ^ number of results
-> TaskOutputType -- ^ type of the result (file of raw)
-> JobInfo
createJobInfoFromConfiguration (ActionConfiguration n _ optEnc _ inEnc _ sConf mConf cConf rConf) opts' ls1 ls2 sCnt mCnt rCnt nor' rt
= JobInfo n opts sa ma ca ra nor i
where
opts = optEnc opts'
sa = maybe Nothing (\_ -> Just $ JobAction n TOTFile sCnt) sConf
ma = maybe Nothing (\_ -> Just $ JobAction n TOTFile mCnt) mConf
ca = maybe Nothing (\_ -> Just $ JobAction n TOTFile mCnt) cConf
ra = maybe Nothing (\_ -> Just $ JobAction n rt rCnt) rConf
nor = Just nor'
i = inEnc ls1 ls2
createListsFromJobResult
:: ActionConfiguration a k1 v1 k2 v2 v3 v4
-> JobResult
-> ([(k2,v4)],[FS.FileId])
createListsFromJobResult ac jr = (ac_OutputDecoder ac) (jr_Output jr)
-- ----------------------------------------------------------------------------
-- SplitAction
-- ----------------------------------------------------------------------------
-- | general SplitAction
type SplitAction a k1 v1 = ActionEnvironment -> a -> Int -> [(k1,v1)] -> IO [(Int, [(k1,v1)])]
-- | SplitAction on ByteStrings
type BinarySplitAction = ActionEnvironment -> B.ByteString -> Maybe Int -> (Int,[FunctionData]) -> IO [(Int, [FunctionData])]
type SplitFunction a k1 v1 = SplitAction a k1 v1
performSplitAction
:: (NFData k1, NFData v1, Binary a, Binary k1, Binary v1, Show k1, Show v1)
=> OptionsDecoder a
-> SplitFunction a k1 v1
-> InputReader k1 v1
-> OutputWriter k1 v1
-> ActionEnvironment
-> B.ByteString
-> Maybe Int
-> (Int,[FunctionData])
-> IO [(Int, [FunctionData])]
performSplitAction optDec fct reader writer env opts n (i,ls)
= do
let a = optDec opts
infoM localLogger "performSplitAction"
putTimeStamp "Begin performSplitAction"
infoM localLogger "reading inputList"
inputList <- readConnector reader env ls
debugM localLogger $ ">>>>>>>>>>>>>>>>>> input is: " ++ show inputList ++ "\n\n"
infoM localLogger "doing split"
partedList <- case n of
(Just n') -> fct env a n' inputList
(Nothing) -> return [(i,inputList)]
debugM localLogger $ ">>>>>>>>>>>>>>>>>> splittet list is: " ++ show partedList ++ "\n\n"
infoM localLogger "writing outputlist"
outputList <- writeConnector writer env partedList
putTimeStamp "End performSplitAction"
return outputList
-- ----------------------------------------------------------------------------
-- MapAction
-- ----------------------------------------------------------------------------
-- | general MapAction
type MapAction a k1 v1 k2 v2 = ActionEnvironment -> a -> Int -> [(k1,v1)] -> IO [(Int, [(k2,v2)])]
-- | MapAction on ByteStrings
type BinaryMapAction = ActionEnvironment -> B.ByteString -> Maybe Int -> (Int,[FunctionData]) -> IO [(Int, [FunctionData])]
type MapFunction a k1 v1 k2 v2 = ActionEnvironment -> a -> k1 -> v1 -> IO [(k2, v2)]
type MapPartition a k2 v2 = ActionEnvironment -> a -> Int -> [(k2,v2)] -> IO [(Int, [(k2,v2)])]
performMapAction
:: (Ord k2, Show k1, Show k2, Show v1, Show v2,
Binary a, Binary k1, Binary v1, Binary k2, Binary v2, NFData k2, NFData v2, NFData v1, NFData k1)
=> OptionsDecoder a
-> MapFunction a k1 v1 k2 v2
-> MapPartition a k2 v2
-> InputReader k1 v1
-> OutputWriter k2 v2
-> ActionEnvironment
-> B.ByteString
-> Maybe Int
-> (Int,[FunctionData])
-> IO [(Int, [FunctionData])]
performMapAction optDec fct part reader writer env opts n (i,ls)
= do
-- decode the options
let a = optDec opts
infoM localLogger "performMapAction"
putTimeStamp "Begin performMapAction"
infoM localLogger $ "reading inputList: " ++ show (i,ls)
inputList <- readConnector reader env ls
debugM localLogger $ ">>>>>>>>>>>>>>>>>> input is: " ++ show inputList ++ "\n\n"
infoM localLogger ("doing map: " ++ (show . length $ inputList))
mappedList <- mapM (\(k1, v1) -> fct env a k1 v1) inputList
let tupleList = concat mappedList
infoM localLogger $ "map result: " ++ (show . length $ tupleList)
debugM localLogger $ ">>>>>>>>>>>>>>>>>> mapped list is: " ++ show tupleList ++ "\n\n"
infoM localLogger "doing partition"
partedList <- case n of
(Just n') -> part env a n' tupleList
(Nothing) -> return [(i,tupleList)]
infoM localLogger $ "map parted result: " ++ (show . length $ partedList)
debugM localLogger $ ">>>>>>>>>>>>>>>>>> partitioned list is: " ++ show partedList ++ "\n\n"
infoM localLogger "writing outputlist: begin"
outputList <- writeConnector writer env partedList
infoM localLogger "writing outputlist: done"
putTimeStamp "End performMapAction"
return outputList
-- ----------------------------------------------------------------------------
-- Combine- / ReduceTask
-- ----------------------------------------------------------------------------
-- | general MapAction
type ReduceAction a k2 v2 v3 = ActionEnvironment -> a -> Int -> [(k2,v2)] -> IO [(Int, [(k2,v3)])]
-- | MapAction on ByteStrings
type BinaryReduceAction = ActionEnvironment -> B.ByteString -> Maybe Int -> (Int,[FunctionData]) -> IO [(Int, [FunctionData])]
type ReduceMerge a k2 v2 = ActionEnvironment -> a -> [(k2,v2)] -> IO [(k2,[v2])]
type ReduceFunction a k2 v2 v3 = ActionEnvironment -> a -> k2 -> [v2] -> IO (Maybe v3)
type ReducePartition a k2 v3 = ActionEnvironment -> a -> Int -> [(k2,v3)] -> IO [(Int, [(k2,v3)])]
performReduceAction
:: (Ord k2, Show k2, Show v2, Show v3,
NFData k2, NFData v2,
Binary a, Binary k2, Binary v2, Binary v3)
=> OptionsDecoder a
-> ReduceMerge a k2 v2
-> ReduceFunction a k2 v2 v3
-> ReducePartition a k2 v3
-> InputReader k2 v2
-> OutputWriter k2 v3
-> ActionEnvironment
-> B.ByteString
-> Maybe Int
-> (Int,[FunctionData])
-> IO [(Int, [FunctionData])]
performReduceAction optDec merge fct part reader writer env opts n (i,ls)
= do
-- decode the options
let a = optDec opts
infoM localLogger "performReduceAction"
putTimeStamp "Begin performReduceAction"
infoM localLogger "reading inputList"
inputList <- readConnector reader env ls
infoM localLogger $ ">>>>>>>>>>>>>>>>>> input is: " ++ (show . length $ inputList) ++ "\n\n"
debugM localLogger $ ">>>>>>>>>>>>>>>>>> input is: " ++ show inputList ++ "\n\n"
infoM localLogger "doing merge"
mergedList <- merge env a inputList
debugM localLogger $ ">>>>>>>>>>>>>>>>>> mergedList is: " ++ show mergedList ++ "\n\n"
infoM localLogger "doing reduce"
maybesList <- mapM (\(k2,v2s) -> performReduceFunction a k2 v2s) mergedList
let tupleList = catMaybes maybesList
debugM localLogger $ ">>>>>>>>>>>>>>>>>> tupleList is: " ++ show tupleList ++ "\n\n"
infoM localLogger "doing partition"
partedList <- case n of
(Just n') -> part env a n' tupleList
(Nothing) -> return [(i,tupleList)]
debugM localLogger $ ">>>>>>>>>>>>>>>>>> partedList is: " ++ show partedList ++ "\n\n"
infoM localLogger "writing outputlist: begin"
outputList <- writeConnector writer env partedList
infoM localLogger "writing outputlist: done"
putTimeStamp "End performReduceAction"
return outputList
where
performReduceFunction a k2 v2s
= do
mbV3 <- fct env a k2 v2s
case mbV3 of
(Nothing) -> return Nothing
(Just v3) -> return $ Just (k2,v3)
putTimeStamp :: String -> IO ()
putTimeStamp s = do
t1 <- getPOSIXTime
infoM localLogger (s++" : "++ show t1)