diff --git a/db/library.js b/db/library.js new file mode 100644 index 0000000..d8e4b0b --- /dev/null +++ b/db/library.js @@ -0,0 +1,224 @@ +// ETL for open library: https://openlibrary.org/developers/dumps + +const { pipeline } = require('node:stream/promises') +const { createGunzip } = require('node:zlib') + +const cds = require('@sap/cds'); + +(async () => { + process.env.CDS_TEST_FAKE = 'true' + await cds.test(cds.utils.path.resolve(__dirname, '..')) + await cds.deploy(cds.options.from) +})() + .then(() => Promise.all([ + store('https://openlibrary.org/data/ol_dump_authors_latest.txt.gz', 'Authors'), + store('https://openlibrary.org/data/ol_dump_works_latest.txt.gz', 'Books'), + ])) + .then(() => { // Create foreign key index + const { Books } = cds.entities + return cds.run(`CREATE INDEX ${Books}_author ON ${Books}(author_ID)`) + }) + .then(() => cds.shutdown()) + +async function store(src, target, maxsize = 1000) { + let transform + switch (target) { + case 'Authors': transform = Authors + break; + case 'Books': transform = Books + break; + default: cds.error`unknown target: ${target}` + } + return pipeline(await open(src), createGunzip(), extract, transform, throughput.bind(target), insert.bind(null, target, maxsize)) +} + +async function* Authors(data) { + let errors = 0 + let success = 0 + for await (const row of data) { + try { + yield { + ID: ID(row.key), + name: row.name.slice(0, 255), + createdAt: timestamp(row.created), + modifiedAt: timestamp(row.last_modified), + } + success++ + } catch { errors++ } + } + console.log(`Transformed ${success} Authors (ignored ${errors})`) +} + +async function* Books(data) { + let errors = 0 + let success = 0 + for await (const row of data) { + try { + yield { + ID: ID(row.key), + title: row.title.slice(0, 255), + author_ID: ID(row.authors[0].author.key), + createdAt: timestamp(row.created), + modifiedAt: timestamp(row.last_modified), + } + success++ + } catch { errors++ } + } + console.log(`Transformed ${success} Books (ignored ${errors})`) +} + +async function insert(target, maxsize, data) { + if (typeof maxsize !== 'number') { + data = maxsize + maxsize = 1000 + } + target = cds.model.entities('sap.capire.bookshop')[target] + let buffer = [] + for await (const row of data) { + if (buffer.length >= maxsize) { + await UPSERT(buffer).into(target).then(() => { }, err => console.log('error:', err.message)) + buffer = [] + } + buffer.push(row) + } + if (buffer.length) await UPSERT(buffer).into(target) +} + +async function write(target, maxsize, data) { + if (typeof maxsize !== 'number') { + data = maxsize + maxsize = 1024 * 1024 * 100 + } + const filename = cds.utils.path.resolve(__dirname, 'data', `sap.capire.bookshop-${target}`) + let file, index = 0, size = 0, sep = '' + for await (const row of data) { + if (!file || size > maxsize) { + if (file) file.end(']') + file = cds.utils.fs.createWriteStream(`${filename}-${index++}.json`) + file.write('[') + size = 1 + sep = '' + } + + const chunk = JSON.stringify(row) + size += sep.length + chunk.length + + if (sep) { file.write(sep) } else { sep = ',' } + file.write(chunk) + } + file.end(']') +} + +async function* extract(data) { + for await (const line of split(data)) { + try { yield JSON.parse(line.split('\t').at(-1)) } catch { } + } +} + +async function* split(stream) { + let leftover = '' + for await (const chunk of stream) { + const split = `${leftover}${chunk}`.split('\n') + for (const line of split) if (line) yield line + } + if (leftover) yield leftover +} + +const { stdout } = process +async function throughput_print() { + const now = Date.now() + + stdout.cursorTo(0, 0) + + const nameSize = Object.keys(throughput.trackedStream).reduce((l, c) => Math.max(l, c.length), 0) + const width = 35 + nameSize + const wrap = '\n'.padStart(width, '=') + stdout.write(wrap) + for (const name in throughput.trackedStream) { + const stats = throughput.trackedStream[name] + if (!stats.start) { + stdout.write(`${name.padStart(nameSize, ' ')}: waiting...\n`) + continue + } + const format = stats.type === 'objects' ? objects : bytes + // stdout.write(`${name.padStart(nameSize, ' ')}: ${format(stats.size / ((now - stats.start) / 1000)).padStart(10, ' ')}/sec total: ${format(stats.size).padStart(10, ' ')} (${time(now - stats.start)})\n`) + stdout.write(`${name.padStart(nameSize, ' ')}: ${stats.size.toLocaleString().padStart(23, ' ')} (${time(now - stats.start)})\n`) + } + stdout.write(wrap) +} + +function bytes(bytes) { + let scale = 0 + while (bytes > 1024) { + bytes /= 1024 + scale++ + } + const scales = [' B', 'KiB', 'MiB', 'GiB', 'TiB'] + return `${(bytes >>> 0).toLocaleString()} ${scales[scale]}` +} + +function objects(count) { + let scale = 0 + while (count > 1000) { + count /= 1000 + scale++ + } + const scales = ['K ', 'K ', 'M ', 'B ', 'T ', 'Q ', 'P '] + return `${(count >>> 0).toLocaleString()} ${scales[scale]}` +} + +function time(milliseconds) { + const scales = ['mil', 'sec', 'min', 'Hur', 'Day', 'Wek', 'Mon', 'Yer'] + const scaleSize = [1000, 60, 60, 24, 7, 29, 256] + let scale = 0 + while (milliseconds > scaleSize[scale]) { + milliseconds /= scaleSize[scale] + scale++ + } + return `${(milliseconds >>> 0).toLocaleString()} ${scales[scale]}` +} + +async function* throughput(name, stream) { + if (typeof name !== 'string') { + stream = name + name = this + } + + const stats = (throughput.trackedStream ??= {})[name] = { start: undefined, size: 0, type: undefined } + for await (const chunk of stream) { + throughput.interval ??= setInterval(throughput_print, stdout.isTTY ? 100 : 10_000) + stats.start ??= Date.now() + stats.type ??= typeof (chunk?.bytelength ?? chunk?.length) === 'number' ? 'bytes' : 'objects' + stats.size += chunk?.bytelength ?? chunk?.length ?? 1 + yield chunk + } + delete throughput.trackedStream[name] + if (!Object.keys(throughput.trackedStream).length) throughput.interval = clearInterval(throughput.interval) +} + +async function open(src) { + const filename = cds.utils.path.resolve(__dirname, 'data', cds.utils.path.basename(src)) + + if (!cds.utils.fs.existsSync(filename)) { + const download = await fetch(src) + if (download.status > 399) cds.error`Failed to download ${filename} (${download.statusText || download.status})` + await pipeline(download.body, cds.utils.fs.createWriteStream(filename)) + } + + return cds.utils.fs.createReadStream(filename) +} + +function timestamp(val) { + if (!val) return null + const value = val.value || val + return value[value.length - 1] === 'Z' ? value : `${value}Z` +} + +function ID(key) { + return Number.parseInt(/\d+/.exec(key)[0]) +} + +module.exports = { + open, extract, write, throughput, + Authors, Books, +}