如何从compojure API流式传输大型CSV响应,以便整个响应不会立即保存在内存中？

Question

我是新手使用compojure,但到目前为止一直喜欢使用它.我目前在我的一个API端点中遇到一个问题,即从数据库生成一个大的CSV文件,然后将其作为响应主体传递.

我似乎遇到的问题是整个CSV文件被保存在内存中,然后导致API中的内存不足错误.处理和生成这个的最佳方法是什么,理想情况下是一个gzip压缩文件？是否可以流式传输响应,以便一次返回几千行？当我为相同的数据返回JSON响应主体时,返回它没有问题.

这是我用来返回的当前代码:

(defn complete
  "Returns metrics for each completed benchmark instance"
  [db-client response-format]
  (let [benchmarks  (completed-benchmark-metrics {} db-client)]
    (case response-format
      :json  (json-grouped-output field-mappings benchmarks)
      :csv   (csv-output benchmarks))))

(defn csv-output [data-seq]
  (let [header (map name (keys (first data-seq)))
        out    (java.io.StringWriter.)
        write  #(csv/write-csv out (list %))]
    (write header)
    (dorun (map (comp write vals) data-seq))
    (.toString out)))

这data-seq是从数据库返回的结果,我认为这是一个懒惰的序列.我正在使用yesql来执行数据库调用.

以下是此API端点的compojure资源:

(defresource results-complete [db]
  :available-media-types  ["application/json" "text/csv"]
  :allowed-methods        [:get]
  :handle-ok              (fn [request]
                            (let [response-format (keyword (get-in request [:request :params :format] :json))
                                  disposition     (str "attachment; filename=\"nucleotides_benchmark_metrics." (name response-format) "\"")
                                  response        {:headers {"Content-Type" (content-types response-format)
                                                             "Content-Disposition" disposition}
                                                   :body    (results/complete db response-format)}]
                              (ring-response response))))

Answer 1

感谢此主题中提供的所有建议,我能够使用piped-input-stream以下方法创建解决方案:

(defn csv-output [data-seq]
  (let [headers     (map name (keys (first data-seq)))
        rows        (map vals data-seq)
        stream-csv  (fn [out] (csv/write-csv out (cons headers rows))
                              (.flush out))]
    (piped-input-stream #(stream-csv (io/make-writer % {})))))

这与我的解决方案不同,因为它没有实现序列使用dorun,也没有创建大String对象.而是按照文档中的描述PipedInputStream异步写入连接:

从以输出流作为参数的函数创建输入流.该函数将在一个单独的线程中执行.函数完成后,流将自动关闭.