如何在使用 pg-query-stream 和 pg-promise 进行流式传输时运行更新？

Question

• 我正在尝试从数据库中加载 50000 个包含文本的项目，标记它们并更新标签
• 我为此目的使用 pg-promise 和 pg-query-stream
• 我能够让流式传输部分正常工作，但由于更新语句太多，更新变得有问题

这是我现有的代码

const QueryStream = require('pg-query-stream')
const JSONStream = require('JSONStream')

function prepareText(title, content, summary) {
  let description
  if (content && content.length) {
    description = content
  } else if (summary && summary.length) {
    description = summary
  } else {
    description = ''
  }
  return title.toLowerCase() + ' ' + description.toLowerCase()
}

async function tagAll({ db, logger, tagger }) {
  // you can also use pgp.as.format(query, values, options)
  // to format queries properly, via pg-promise;
  const qs = new QueryStream(
    'SELECT feed_item_id,title,summary,content FROM feed_items ORDER BY pubdate DESC, feed_item_id DESC'
  )
  try {
    const result = await db.stream(qs, (s) => {
      // initiate streaming into the console:
      s.pipe(JSONStream.stringify())
      s.on('data', async (item) => {
        try {
          s.pause()
          // eslint-disable-next-line camelcase
          const { feed_item_id, title, summary, content } = item

          // Process text to be tagged
          const text = prepareText(title, summary, content)
          const tags = tagger.tag(text)

          // Update tags per post
          await db.query(
            'UPDATE feed_items SET tags=$1 WHERE feed_item_id=$2',
            // eslint-disable-next-line camelcase
            [tags, feed_item_id]
          )
        } catch (error) {
          logger.error(error)
        } finally {
          s.resume()
        }
      })
    })
    logger.info(
      'Total rows processed:',
      result.processed,
      'Duration in milliseconds:',
      result.duration
    )
  } catch (error) {
    logger.error(error)
  }
}

module.exports = tagAll

• db 对象是来自 pg-promise 的对象，而标记器只是从变量 tags 中包含的文本中提取标签数组
• 根据我在诊断中看到的情况，正在执行太多更新语句，有没有办法对它们进行批处理？

Answer 1

这是我能想到的最好的方法来批处理流中的查询，这样我们就不需要在内存中加载所有数据或运行太多查询。如果有人知道更好的批处理方法，尤其是使用 t.sequence 进行批处理，请随时添加另一个答案

const BATCH_SIZE = 5000
async function batchInsert({ db, pgp, logger, data }) {
  try {
    // https://vitaly-t.github.io/pg-promise/helpers.ColumnSet.html
    const cs = new pgp.helpers.ColumnSet(
      [
        { name: 'feed_item_id', cast: 'uuid' },
        { name: 'tags', cast: 'varchar(64)[]' },
      ],
      {
        table: 'feed_items',
      }
    )
    const query =
      pgp.helpers.update(data, cs) + ' WHERE v.feed_item_id=t.feed_item_id'
    await db.none(query)
  } catch (error) {
    logger.error(error)
  }
}

async function tagAll({ db, pgp, logger, tagger }) {
  // you can also use pgp.as.format(query, values, options)
  // to format queries properly, via pg-promise;
  const qs = new QueryStream(
    'SELECT feed_item_id,title,summary,content FROM feed_items ORDER BY pubdate DESC, feed_item_id DESC'
  )
  try {
    const queryValues = []
    const result = await db.stream(qs, (s) => {
      // initiate streaming into the console:
      s.pipe(JSONStream.stringify())
      s.on('data', async (item) => {
        try {
          s.pause()
          // eslint-disable-next-line camelcase
          const { feed_item_id, title, summary, content } = item

          // Process text to be tagged
          const text = prepareText(title, summary, content)
          const tags = tagger.tag(text)
          queryValues.push({ feed_item_id, tags })

          if (queryValues.length >= BATCH_SIZE) {
            const data = queryValues.splice(0, queryValues.length)
            await batchInsert({ db, pgp, logger, data })
          }
        } catch (error) {
          logger.error(error)
        } finally {
          s.resume()
        }
      })
    })
    await batchInsert({ db, pgp, logger, data: queryValues })
    return result
  } catch (error) {
    logger.error(error)
  }
}

Answer 2

如果您可以使用一条 sql 语句完成所有操作，那么您应该这样做！在这里，您要为表的每一行在节点和数据库之间来回付出代价，这将占用您查询的大部分时间。

你的请求可以用纯sql实现：

update feed_items set tags=case 
    when (content = '') is false then lower(title) || ' ' || lower(content) 
    when (summary = '') is  false then lower(title) || ' ' || lower(summary) 
    else title end;

此请求将立即更新您的所有表格。我相信它会比你的方法快一些数量级。在我的机器上，一张包含 100000 行的表，更新时间约为 600ms。

一些备注：

• 您不需要订购更新。由于订购速度很慢，最好不要这样做。
• 我猜limit 部分是因为它太慢了？如果是这种情况，那么你可以放弃它，50000 行对于 postgres 来说不是一张大表。
• 我敢打赌，pg-stream 的事情确实不是真的从数据库中流出东西，它只允许您使用它之前收集的结果中的类似流的 api... 没问题那个，但我认为这里可能存在误解。