流式处理xml-conduit解析结果

Question

我想使用xml-conduit,特别Text.XML.Stream.Parse是为了从大型XML文件中懒惰地提取对象列表.

作为测试用例,我使用最近重新发布的StackOverflow数据转储.为了简单起见,我打算从中提取所有用户名stackoverflow.com-Users.7z.即使文件是a .7z,file说它只是bzip2压缩数据(文件末尾可能有一些7zip的东西,但是现在我不在乎).

XML的简化版本将是

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

基于此之前的Q&A以及实例上Hackage流读取示例XML在BZ2-ED形式的作品完美的我

但是,在使用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).

我如何将结果"直播"到stdout？我假设这是可能的,通过添加另一个管道语句,如$$ CB.sinkFile "output.txt"最后.然而,这个特定版本需要Conduit输出ByteString.你能指出我正确的方向从哪里开始吗？

任何帮助将不胜感激!

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).

我已根据您的代码整理了一个完整的示例.输入是一个人工生成的无限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

Answer 2

进行了编辑，以更新M. Snoyman的富有洞察力的示例，但它被平庸的权力绊倒者扔掉了。因此，这个。

\n\n

原始版本将不再编译并产生许多已弃用的警告（旧语法）。

\n\n

{-# LANGUAGE OverloadedStrings #-}\n{-# LANGUAGE RankNTypes        #-}\nimport           Control.Applicative    ((<*))\nimport           Control.Concurrent     (threadDelay)\nimport           Control.Monad          (forever, void)\nimport           Control.Monad.Catch    (MonadThrow)\nimport           Control.Monad.IO.Class (MonadIO (liftIO))\nimport           Data.ByteString        (ByteString)\nimport           Data.Conduit\nimport qualified Data.Conduit.List      as CL\nimport           Data.Text              (Text)\nimport           Data.Text.Encoding     (encodeUtf8)\nimport           Data.XML.Types         (Event)\nimport           Text.XML.Stream.Parse\n\n-- instead of actually including a large input data file, just for testing purposes\ninfiniteInput :: MonadIO m => ConduitT () ByteString m ()\ninfiniteInput = do\n    yield "<users>"\n    forever $ do\n        yield $ encodeUtf8\n            "<row id=\\"1\\" DisplayName=\\"StackOverflow\\"/><row id=\\"2597135\\" DisplayName=\\"Uli K\xc3\xb6hler\\"/>"\n        liftIO $ threadDelay 1000000\n    --yield "</users>" -- will never be reached\n\ndata User = User {name :: Text} deriving (Show)\n\nparseUserRow :: MonadThrow m => forall o. ConduitT Event o m (Maybe User)\nparseUserRow = tag\' "row" (requireAttr "DisplayName" <* ignoreAttrs) $ \\displayName -> do\n    return $ User displayName\n\nparseUsers :: MonadThrow m => ConduitT Event User m ()\nparseUsers = void $ tagNoAttr "users" $ manyYield parseUserRow\n\n--or use manyYield, now provided by Text.XML.Stream.Parse\nyieldWhileJust :: Monad m\n               => ConduitT a b m (Maybe b)\n               -> ConduitT a b m ()\nyieldWhileJust consumer =\n    loop\n  where\n    loop = do\n        mx <- consumer\n        case mx of\n            Nothing -> return ()\n            Just x -> yield x >> loop\n\nmain :: IO ()\nmain = runConduit $ infiniteInput\n    .| parseBytes def\n    .| parseUsers\n    .| CL.mapM_ print\n

\n\n

ghc 8.6.5，xml-conduit 1.9.0.0

\n