xml - 流式传输 xml-conduit 解析结果

Question

我想使用 xml-conduit , 具体来说 Text.XML.Stream.Parse为了从大型 XML 文件中延迟提取对象列表。

作为测试用例，我使用 the recently re-released StackOverflow data dumps .为简单起见，我打算从 stackoverflow.com-Users.7z 中提取所有用户名。即使文件是 .7z，file 也表示它只是 bzip2 压缩数据(文件末尾可能有一些 7zip 的东西，但现在我不在乎)。

XML 的简化版本是

<users>
    <row id="1" DisplayName="StackOverflow"/>
    ...
    <row id="2597135" DisplayName="Uli Köhler"/>
    ... 
</users>

基于 this previous Q&A和示例 on Hackage流式读取 bz2-ed 形式的示例 XML 非常适合我

但是，当使用 runghc 运行以下程序时，它运行时没有打印任何输出:

{-# LANGUAGE OverloadedStrings #-}
import Data.Conduit (runResourceT, ($$), ($=))
import qualified Data.Conduit.Binary as CB
import Data.Conduit.BZlib
import Data.Conduit
import Data.Text (Text)
import System.IO
import Text.XML.Stream.Parse
import Control.Applicative ((<*))

data User = User {name :: Text} deriving (Show)

parseUserRow = tagName "row" (requireAttr "DisplayName" <* ignoreAttrs) $ \displayName -> do
    return $ User displayName

parseUsers = tagNoAttr "users" $ many parseUserRow

main = do
    users <- runResourceT $ CB.sourceFile "stackoverflow.com-Users.7z" $= bunzip2 $= parseBytes def $$ force "users required" parseUsers
    putStrLn $ unlines $ map show users

我认为这个问题的发生是因为 Haskell 试图在开始打印之前深入评估 users 列表。程序的内存使用量以每秒约 2% 的速度持续增长(来源:htop)支持这一理论。

如何将结果“直播”到标准输出？我认为这可以通过在末尾添加另一个管道语句(如 $$ CB.sinkFile "output.txt")来实现。但是，此特定版本需要 ByteString 的 Conduit 输出。你能为我指出正确的方向吗？

任何帮助将不胜感激!

Answer 1

首先让我说 xml-conduit 中的流式助手 API 已经多年未开发，并且可能会受益于在此期间管道发生的给定变化的重新构想。我认为可能有更好的方法来完成任务。

也就是说，让我解释一下您遇到的问题。 many 函数创建一个结果列表，并且在完成处理之前不会产生任何值。在你的情况下，有太多的值，这似乎永远不会发生。最终，当整个文件被读取后，整个用户列表将立即显示。但这显然不是您要寻找的行为。

相反，您要做的是创建 User 值的流，这些值会在准备就绪后立即生成。您要做的基本上是用一个新函数替换 many 函数调用，该函数每次解析时都会 yield 一个结果。一个简单的实现可以是:

yieldWhileJust :: Monad m
               => ConduitM a b m (Maybe b)
               -> Conduit a m b
yieldWhileJust consumer =
    loop
  where
    loop = do
        mx <- consumer
        case mx of
            Nothing -> return ()
            Just x -> yield x >> loop

此外，您不想使用 putStrLn $ unlines $ map show，而是希望将整个管道附加到一个消费者，消费者将打印每个单独产生的 User 值。这可以通过 Data.Conduit.List.mapM_ 轻松实现，例如:CL.mapM_ (liftIO . print)。

我整理了a full example根据你的代码。输入是一个人工生成的无限 XML 文件，只是为了证明它确实是立即产生输出这一点。

{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE RankNTypes        #-}
import           Control.Applicative    ((<*))
import           Control.Concurrent     (threadDelay)
import           Control.Monad          (forever, void)
import           Control.Monad.IO.Class (MonadIO (liftIO))
import           Data.ByteString        (ByteString)
import           Data.Conduit
import qualified Data.Conduit.List      as CL
import           Data.Text              (Text)
import           Data.Text.Encoding     (encodeUtf8)
import           Data.XML.Types         (Event)
import           Text.XML.Stream.Parse

-- instead of actually including a large input data file, just for testing purposes
infiniteInput :: MonadIO m => Source m ByteString
infiniteInput = do
    yield "<users>"
    forever $ do
        yield $ encodeUtf8
            "<row id=\"1\" DisplayName=\"StackOverflow\"/><row id=\"2597135\" DisplayName=\"Uli Köhler\"/>"
        liftIO $ threadDelay 1000000
    --yield "</users>" -- will never be reached

data User = User {name :: Text} deriving (Show)

parseUserRow :: MonadThrow m => Consumer Event m (Maybe User)
parseUserRow = tagName "row" (requireAttr "DisplayName" <* ignoreAttrs) $ \displayName -> do
    return $ User displayName

parseUsers :: MonadThrow m => Conduit Event m User
parseUsers = void $ tagNoAttr "users" $ yieldWhileJust parseUserRow

yieldWhileJust :: Monad m
               => ConduitM a b m (Maybe b)
               -> Conduit a m b
yieldWhileJust consumer =
    loop
  where
    loop = do
        mx <- consumer
        case mx of
            Nothing -> return ()
            Just x -> yield x >> loop

main :: IO ()
main = infiniteInput
    $$ parseBytes def
    =$ parseUsers
    =$ CL.mapM_ print