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 -> b) 类型的并行、本地“映射”作业分配给不同的 spark 来分解像 ([a] -> c) 这样的列表，然后折叠所有导致减少工作为([b] -> c)。通常不保证reduce 步骤的顺序，但RWH 的mapReduce 实际上确实给了我们这样的保证。

RWH 的lineChunks 函数实际上将文档 (LB.ByteString -> [LB.ByteString]) 拆分为少量行的块。我们的映射作业每个都获得这些块之一，并且需要在本地提供有关行匹配的信息。我们可以通过将块拆分为其组成行，本地对行编号，将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) 类型。酷，让我们试着做到这一点。如果我们有Ints 的列表，我们可以重建实际的行号...

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

等等，事实上，我们不能。我们需要在映射器中添加更多信息——每个块中出现的行数。幸运的是，由于我们小心翼翼地让我们的东西解开，这很容易添加。我们需要一个更像([Int], Int) 的返回类型，其中第二个参数是该块的行数。我们可以使用“fanout”(&&&) 来做到这一点。

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

现在我们的输出看起来像

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

我们可以使用State monad 实现一个只计算的reducer

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%。