stream - 使用流时如何限制内存使用

Question

我正在处理一个太大而无法放入内存的数据集，因此我想使用流来处理数据。但是，我发现流并没有像我预期的那样限制使用的内存量。示例代码说明了这个问题，如果内存限制设置为 128mb，将会用完内存。我知道我可以提高内存限制，但是对于我想使用的那种大数据集，这不是一个选择。我应该如何降低内存使用量？

#lang racket

(struct prospects (src pos num max-num)
  #:methods gen:stream
  [(define (stream-empty? stream)
     (equal? (prospects-num stream) (prospects-max-num stream)))
   ;
   (define (stream-first stream)
     (list-ref (prospects-src stream) (prospects-pos stream)))
   ;
   (define (stream-rest stream)
     (let ([next-pos (add1 (prospects-pos stream))]
           [src (prospects-src stream)]
           [next-num (add1 (prospects-num stream))]
           [max-num (prospects-max-num stream)])
       (if (< next-pos (length src))
           (prospects src next-pos next-num max-num)
           (prospects src 0 next-num max-num))))])

(define (make-prospects src num)
  (prospects src 0 0 num))

(define (calc-stats prospects)
  (let ([just-a (stream-filter
                 (λ (p) (equal? "a" (vector-ref p 0)))
                 prospects)]
        [just-b (stream-filter
                 (λ (p) (equal? "b" (vector-ref p 0)))
                 prospects)])
    ;
    (let ([num-a (stream-length just-a)]
          [num-b (stream-length just-b)]
          [sum-ref1-a (for/sum ([p (in-stream just-a)])
                        (vector-ref p 1))]
          [sum-ref1-b (for/sum ([p (in-stream just-b)])
                        (vector-ref p 1))])
      ;
      #|
          ; Have also tried with stream-fold instead of for/sum as below:
          [sum-ref1-a (stream-fold
                       (λ (acc p) (+ acc (vector-ref p 1)))
                       0 just-a)]
          [sum-ref1-b (stream-fold
                       (λ (acc p) (+ acc (vector-ref p 1)))
                       0 just-b)])
      |#
      ;      
      (list num-a num-b sum-ref1-a sum-ref1-b))))

;================================
;           Main
;================================
(define num-prospects 800000)

(define raw-prospects (list #("a" 2 2 5 4 5 6 2 4 2 45 6 2 4 5 6 3 4 5 2)
                            #("b" 1 3 5 2 4 3 2 4 5 34 3 4 5 3 2 4 5 6 3)))

(calc-stats (make-prospects raw-prospects num-prospects))

注意:创建此程序只是为了演示问题；真正的流将访问数据库以引入数据。

Answer 1

您代码中的主要问题是您试图多次通过流。 (每次调用 stream-length 都是一次传递，而你每次调用 for/sum(或 stream-fold，就此而言) 是另一个过程。)这意味着您必须具体化整个流，而不允许更早的流元素被垃圾收集。

这里是对您的代码的修改，只进行一次传递。请注意，我在我的版本中将 num-prospects 设置为 8,000,000，因为即使是多遍版本也没有用完我系统上只有 800,000 的内存:

#lang racket
(require srfi/41)

(define (make-prospects src num)
  (stream-take num (apply stream-constant src)))

(define (calc-stats prospects)
  (define default (const '(0 . 0)))
  (define ht (for/fold ([ht #hash()])
                       ([p (in-stream prospects)])
               (hash-update ht (vector-ref p 0)
                            (λ (v)
                              (cons (add1 (car v))
                                    (+ (cdr v) (vector-ref p 1))))
                            default)))
  (define stats-a (hash-ref ht "a" default))
  (define stats-b (hash-ref ht "b" default))
  (list (car stats-a) (car stats-b) (cdr stats-a) (cdr stats-b)))

;================================
;           Main
;================================
(define num-prospects 8000000)

(define raw-prospects '(#("a" 2 2 5 4 5 6 2 4 2 45 6 2 4 5 6 3 4 5 2)
                        #("b" 1 3 5 2 4 3 2 4 5 34 3 4 5 3 2 4 5 6 3)))

(calc-stats (make-prospects raw-prospects num-prospects))

我应该澄清一下，使用 srfi/41 只是为了能够编写更高效的 make-prospects 版本(不过， 的引用实现>stream-constant 不是很有效，但仍然比您的 prospects 流生成器更有效)； calc-stats 不使用它。