在 Racket 中将 SQL 查询作为惰性流处理答案

【问题标题】：Processing SQL Queries as Lazy Streams in Racket在 Racket 中将 SQL 查询作为惰性流处理
【发布时间】：2020-09-16 01:19:21
【问题描述】：

语言：Racket（带有 SQL 查询代码/指针）库：数据库、球拍/流、球拍/序列

目标：使用 Racket 中的流懒惰地处理 sql 查询的值。

问题一：如何在 Racket 中操作 SQL 查询流对象？（我可以获取流的流优先值，但不能获取流的其余部分！）

#lang racket/base
(require db
         racket/sequence
         racket/stream)

(define db_sql_local
  (mysql-connect
   #:user "<my-username>"
   #:database "<my-database>"
   #:server "<my-server>"
   #:port <my-port>
   #:password "<my-password>"))

;; PROBLEM 1 HERE
(define test-stream
  (sequence->stream
   (in-query
    chembl_sql_local
    "SELECT * FROM <table-name>"
    #:fetch +inf.0)))

(stream-first test-stream)

;; stream-first of test-stream returns the first-row of the table as a '(#vector).

任何建议或 cmets 将不胜感激 - 谢谢！

【问题讨论】：

标签： sql racket etl lazy-sequences

【解决方案1】：

首先in-query返回的序列不包含向量；序列的每个“元素”都包含多个值，每列返回一个。请参阅the Sequence docs 中以“序列的单个元素...”开头的段落，了解多值元素。

其次，使用#:fetch +inf.0（默认行为）意味着在返回序列之前获取所有行。所以上面的代码没有什么懒惰的；您可以改用 query-rows 并获得一个列表，这样会更容易使用（query-rows 确实将每一行表示为一个向量）。

最后，使用stream-rest 获取流的其余部分。例如：

(require db racket/stream racket/sequence)
(define c (sqlite3-connect #:database 'memory))
(define qseq (in-query c "SELECT 1, 2 UNION SELECT 3, 4" #:fetch 1))
qseq
;; => #<sequence>
(define qstream (sequence->stream qseq))
qstream
;; => #<stream>
(stream-first qstream)
;; => 1 2
(stream-rest qstream)
;; => #<stream>
(stream-first (stream-rest qstream))
;; => 3 4

【讨论】：

【解决方案2】：

感谢您的快速回复。 #:fetch 1 参数绝对是我想要让它变得懒惰的东西。我附上了更新的代码，应该延迟流式传输 sql 查询以导出 tsv 文件。

(define sql_server
  (mysql-connect
   #:user <username>
   #:database <db-name>
   #:server <server>
   #:port <port-num>
   #:password <password>))

(define query-->stream
  (lambda (db-conn query)
    (sequence->stream
     (in-query
      db-conn
      query
      #:fetch 1))))

(define print-table-row-to-tsv
  (lambda (ls port)
    (cond
      ((null? ls)
       (fprintf port "~c" #\newline)
       (void))
      ((sql-null? (car ls))
       (fprintf port "~a~c" "NULL" #\tab)
       (print-table-row-to-tsv (cdr ls) port))
      ((null? (cdr ls))
       (fprintf port "~a" (car ls))
       (print-table-row-to-tsv (cdr ls) port))
      (else
       (fprintf port "~a~c" (car ls) #\tab)
       (print-table-row-to-tsv (cdr ls) port)))))

(define get-table-col-names
  (lambda (db-conn tbl-name)
    (map (lambda (x) (vector-ref x 0))
       (query-rows db-conn (string-append "DESCRIBE " tbl-name)))))

(define export-query-result-to-tsv
  (lambda (db-conn tbl-name query)
    (let* ((tbl-col-names (get-table-col-names db-conn tbl-name))
           (output-file (open-output-file (format "~achembl_~a_table.tsv" (find-system-path 'home-dir) tbl-name) #:exists 'replace))        
           (stream (query-->stream db-conn query)))
      (begin
        (print-table-row-to-tsv tbl-col-names output-file)
        (process-stream-to-tsv stream output-file)
        (close-output-port output-file)))))

(define process-stream-to-tsv
  (lambda (stream port)
    (cond
      ((stream-empty? stream)
       (void))
      (else
       (begin
         (print-table-row-to-tsv (call-with-values (lambda () (stream-first stream)) list) port)
         (process-stream-to-tsv (stream-rest stream) port))))))


(export-query-result-to-tsv sql_server "<table-name>" "SELECT * FROM <table-name>;")

【讨论】：