Haskell:使用 MapReduce 搜索子字符串？

Question

我正在尝试使用现有的 MapReduce 实现(Real World Haskell 中的实现)编写一个简单的程序。

作为使用该框架的示例，下面是一些计算文件中单词数的代码:

module Main where

import Control.Monad (forM_)
import Data.Int (Int64)
import qualified Data.ByteString.Lazy.Char8 as LB
import System.Environment (getArgs)

import LineChunks (chunkedReadWith)
import MapReduce (mapReduce, rdeepseq)

wordCount :: [LB.ByteString] -> Int
wordCount = mapReduce rdeepseq (length . LB.words)
                      rdeepseq sum

main :: IO ()
main = do
  args <- getArgs
  forM_ args $ \path -> do
    numWords <- chunkedReadWith wordCount path
    putStrLn $ "Words: " ++ show numWords

我需要使用相同的 MapReduce 框架来编写一个程序来搜索某些字符串(例如“il”)，并返回找到它们的行号。例如，输出可能是:

verILy: found on lines 34, 67, 23
ILlinois: found on lines 1, 2, 56
vILla: found on lines 4, 5, 6

(“il”不需要大写。)

我是 Haskell 初学者，还没有任何具体的想法。我确实注意到 Data.ByteString.Lazy.Char8 类有一个成员函数 findIndices。这个可以用吗？

任何正确方向的代码或提示将不胜感激。

Answer 1

好吧，让我们开始吧。

<小时/>

首先，我们需要一种在字符串中查找单词的方法。这可以使用正则表达式来完成，假设 regex-tdfa 包裹。 Haskell 正则表达式包很好，但非常通用，一开始有点难以阅读。我们将创建一个函数，它只是匹配运算符 (=~) 的包装器主要是为了使类型具体化。

wordHere :: LB.ByteString -> LB.ByteString -> Bool
wordHere word string = string =~ word
-- a.k.a.            = flip (=~)

<小时/>

现在，mapReduce分解一个列表，如 ([a] -> c)通过提供大量并行、本地“映射”作业 (a -> b)到不同的 Spark ，然后将reduce作业中的所有结果折叠为([b] -> c) 。一般来说，不保证reduce步骤的顺序，但是RWH的mapReduce实际上确实给了我们这样的保证。

RWH 的 lineChunks实际上，函数将文档 (LB.ByteString -> [LB.ByteString]) 拆分为少量行的 block 。我们的映射作业每个都会获取这些 block 之一，并且需要在本地提供有关线匹配的信息。我们可以通过将 block 分割成它们的组成行，本地对行进行编号，映射 wordHere 来做到这一点。覆盖它们，并收集本地线路号码，其中 wordHere返回 true。我们首先一般性地进行操作，替换 wordHere与任何谓词 p :: (LB.ByteString -> Bool)

import Control.Arrow

localLinesTrue :: (LB.ByteString -> Bool) -> [LB.ByteString] -> [Int]
localLinesTrue p ls = map fst . filter snd . map (second p) . zip [1..]

现在我们可以创建本地映射器，例如 localLinesTrue (wordHere $ LB.pack "foobar") . LB.lines :: LB.ByteString -> [Int] 。

给定该类型的映射器还稍微阐明了函数其余部分的类型。我们现在有

>>> :t mapReduce rdeepseq (localLinesTrue (wordHere "foo")) rdeepseq
...    :: ([[Int]] -> c) -> [LB.ByteString] -> c

所以我们的 reducer 必须是 ([[Int]] -> c) 类型。太酷了，让我们尝试一下吧。如果我们有一个列表 Int我们可以重建实际的行号...

[[], [], [], [5], [3], [], []]

等等，实际上，我们不能。我们需要向映射器中添加更多信息——每个 block 中出现的行数。幸运的是，由于我们小心翼翼地将我们的东西分开，因此很容易添加。我们需要一个更像 ([Int], Int) 的返回类型其中第二个参数是该 block 的行数。我们可以通过“扇出”(&&&)来做到这一点.

mapper regex = (localLinesTrue (wordHere regex) &&& length) . LB.lines

现在我们的输出将如下所示

[ ([], 3),  ([], 5),  ([3, 5], 10), ... ]

我们可以实现一个仅使用 State 进行计数的 reducer 单子(monad)

reducerStep :: ([Int], Int) -> State Int [Int]
reducerStep (hits, count) = do pos <- get
                               modify (+count)
                               return (map (+pos) hits)

reducer :: [([Int], Int)] -> [Int]
reducer = concat . evalState 0 . mapM reducerStep

我们已经有了

mapReduce rdeepseq (mapper regex)
          rdeepseq reducer
  :: [LB.ByteString] -> [Int]

这应该能让你完成 95% 的任务。