【问题标题】:Incorporating Conduit to ordinary functions将 Conduit 纳入普通功能
【发布时间】:2014-02-12 12:34:17
【问题描述】:

我编写了一个简单的程序,我在其中读取了一个大的 XML 文件并执行了一些操作 处理文件的内容,然后保存处理的 新文件中的数据。

原来的 main 函数如下:

main = do
  content <- B.readFile "/home/sibi/github/data/chennai.osm" 
  let tags = removeUnwanted $ parseTags content 
      hospitals = toHospital $ extractHospitalNode tags
  BL.writeFile "osmHospitals.json" (encode hospitals)

但是这段代码会占用全部内存并且需要很长时间才能完成。所以,我决定 使用管道库使程序在恒定内存中运行。

但是看了conduit tutorial之后,我还是没明白 如何使上述程序使用管道库的特性。

我发现我可以使用管道的sourceFile,它可以流式传输 文件的内容。但是接下来如何应用函数parseTags(这是一个来自TagSoup库的函数)和其他简单的函数 现在到流媒体内容?

编辑:整个代码是here

【问题讨论】:

    标签: haskell stream conduit


    【解决方案1】:

    parseTags 的方法与 conduitpipes 的方法之间存在巨大的脱节:parseTags 假设它可以纯粹访问下一个数据块,而 pipes/conduit 让您处理不可能的情况,例如从文件流式传输。为了将解析混合到 pipes/conduit 中,您必须有一种方法可以将使用解析混合到提取新数据块的步骤中。

    (我将在续集中使用pipes,因为我更熟悉它们,但这个想法是可以转移的。)

    我们可以在类型中看到这种脱节,尽管我将从一个稍微受限的版本开始。

    parseTags :: Lazy.ByteString -> [Tag Lazy.ByteString]
    

    我们可以将Lazy.ByteString 本身视为流媒体设备,毕竟它本质上只是

    type LazyByteString = [Strict.ByteString]
    

    这样,如果我们自己生成Lazy.ByteString,那么我们可以依靠列表的惰性来确保我们生成的数量不会超过parseTags 继续进行所需的数量(我假设,不看, parseTags 被写入以便它可以增量解析这样的流结构)。

    sillyGen :: LazyByteString
    sillyGen = gen 10 where
      gen 0 = []
      gen n = "<tag> </tag>" : gen (n-1)
    

    现在这里的问题是列表的流式传输行为关键取决于能够纯粹生成列表的尾部。在到目前为止的讨论中,根本没有提到任何单子。不幸的是,对于从文件中流式传输的字符串,情况并非如此——我们需要以某种方式在每个流式传输块之间集成一个 IO 操作,在此我们考虑是否已达到 EOF 并在必要时关闭文件。

    这正是pipesconduit 的领域,所以让我们看看解决这个问题的方法。

    -- from pipes-bytestring
    fromHandle :: Handle -> Producer' Strict.ByteString IO ()
    

    我们可以将fromHandle 视为等同于

    的“一元交织”
    Lazy.hGetContents :: Handle -> IO Lazy.ByteString
    

    这些类型表明了这两个操作之间的关键区别——hGetContents 可以在一个 IO 操作中执行,而当我们将 Handle 传递给 pipes-bytestringfromHandle 时,它返回一个类型在IO 上参数化,但不能简单地从中释放。这完全表明 hGetContents 使用惰性 IO(由于使用 unsafeInterleaveIO 而可能无法预测),而 fromHandle 使用确定性流。

    我们可以写一个类似于Producer Strict.ByteString IO ()的类型为

    data IOStreamBS = IOSBS { stepStream :: IO (Strict.ByteString, Either IOStreamBS ()) }
    

    换句话说,我们可以认为Producer Strict.ByteString IO () 只不过是一个 IO 操作,它准确地生成文件的下一个块,并且(可能)一个新的操作来获取下一个块。这就是pipesconduit 提供确定性流的方式。

    但这也意味着你不能一举逃离IO——你必须随身携带。


    因此,我们可能想要调整parseTags,它能够对其输入进行一些泛化,只接受Producer Strict.ByteString IO () 作为StringLike 类型

    parseTags :: StringLike str => str -> [Tag str]
    

    假设我们已经实例化了StringLike (Producer Strict.ByteString IO ())。这意味着将parseTags 应用于我们的生产者将为我们提供Tag (Producer Strict.ByteString IO ()) 的列表。

    type DetStream = Producer Strict.ByteString IO ()
    parseTags :: DetStream -> [Tag DetStream]
    

    为了实现这一点,我们必须查看 Producer 并将其分割成块,而不在 IO monad 中执行任何操作。至此,应该清楚这样的功能是不可能的——我们甚至无法从文件中获取第一个块,而无需在IO 中执行某些操作。


    为了解决这种情况,出现了像 pipes-parsepipes-group 这样的系统,它们将函数签名替换为类似的东西

    parseTagsGrouped :: Producer Strict.ByteString IO () 
                     -> FreeT (Producer (Tag Strict.ByteString) IO) IO ()
    

    这看起来很吓人,但与parseTags 的用途相同,除了它将列表概括为允许我们在每个元素之间执行任意IO 操作的结构。如类型所示,这种转换可以纯粹地完成,因此允许我们使用纯粹的组合来组装我们的流式机器,并且在最后执行它时只会产生一个IO 步骤(使用runEffect)。


    所以,总而言之,可能无法使用pipesconduit 流式传输到parseTags——它只是假设某些转换可以纯粹完成,推送所有@ 987654378@ 到一个时间点,而pipes/conduit 基本上是在整个计算过程中传播IO 的机制,而不需要太多的脑力开销。

    但是,如果您在使用 parseTags 时遇到问题,只要小心,您可以使用惰性 IO 来解决问题。尝试使用来自Data.ByteString.LazyhGetContents 的一些变化。主要问题是文件可能在unsafeInterleaveIO'd 操作实际开始读取它之前关闭。因此,您需要非常谨慎地管理严格性。

    基本上这就是pipes/conduit 和惰性IO 之间的最大区别。当使用惰性 IO 时,所有的“读取块”操作都是不可见的,并由 Haskell 惰性隐式控制。这是动态的、隐含的、难以观察或预测的。在pipes/conduit 中,所有这些动作都非常明确和静态,但复杂性由您来管理。

    【讨论】:

      【解决方案2】:

      如果你尝试System.IO,逐行读取文件并处理它(或读取部分xml文件)怎么办?

      【讨论】:

      • 懒惰的评估不会起作用吗?我认为使用导管或管道库是解决此问题的更好方法。
      • 那只是一个“如果......”。惰性评估在管道中很棒,但我不喜欢管道。看起来很丑。
      猜你喜欢
      • 2017-04-05
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      • 2012-03-03
      • 1970-01-01
      • 1970-01-01
      • 1970-01-01
      相关资源
      最近更新 更多