Обработка SQL-запросов как ленивых потоков в Racket

Язык: Racket (с кодом/указателем SQL-запроса) Библиотеки: db, racket/stream, racket/sequence

Цель: лениво обрабатывать значение sql-запросов, используя потоки в Racket.

Вопрос 1: как вы манипулируете объектами потока запросов SQL в Racket? (Я могу получить первое значение потока для потока, но не для остального потока!)

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

Любые советы или комментарии будут с благодарностью - Спасибо!


sql etl lazy-sequences racket
person Triage    schedule 16.09.2020    source источник

Ответы (2)


arrow_upward
2
arrow_downward

Во-первых, последовательность, возвращаемая in-query, не содержит векторов; каждый элемент последовательности содержит несколько значений, по одному на возвращаемый столбец. См. абзац в документах по последовательности, начиная с отдельных элементов последовательности... о многозначных элементах.

Во-вторых, использование #: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
person Ryan Culpepper    schedule 16.09.2020

arrow_upward
1
arrow_downward

Спасибо за ваш быстрый ответ. Аргумент #: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>;")
person Triage    schedule 17.09.2020