如何规避错误 pq_flush: could not send data to client: Broken pipe found

Posted 2023-03-31

【中文标题】如何规避错误 pq_flush: could not send data to client: Broken pipe found

【英文标题】：How to circumvent error pq_flush: could not send data to client: Broken pipe found

【发布时间】：2021-02-11 22:24:09

【问题描述】：

我正在尝试使用看起来像这样的 Lambda 函数创建一个异步进程：

Lambda 1 在我的 redshift 集群上触发查询并结束 Lambda 2 轮询集群以获取查询状态，并根据结果结束/成功

我使用了几种不同的选项，但它们似乎都在某一时刻失败了。我可以创建一个查询并触发它，并让 lambda 结束，但是当查询完成执行而不是成功时，它会抱怨客户端连接不再存在

error pq_flush: could not send data to client: Broken pipe found in xyz

问题是这完全符合我的用例。我不希望客户端（Lambda 1）等待，因为我的查询可能需要一个小时才能运行（夸大但可能），这就是我创建第二个 lambda 的原因。有没有办法可以将它传达给 Redshift/postgresql 并规避这个问题？

这是我的触发代码（最终会转到 lambda，但我正在本地机器上进行测试）

import select
import psycopg2

def wait(conn):
  while True:
    state = conn.poll()
    if state == psycopg2.extensions.POLL_OK:
      break
    elif state == psycopg2.extensions.POLL_WRITE:
      select.select([], [conn.fileno()], [])
    elif state == psycopg2.extensions.POLL_READ:
      select.select([conn.fileno()], [], [])
    else:
      raise psycopg2.OperationalError("poll() returned %s" % state)

conn = psycopg2.connect(
  user='someuser',
  dbname='somedb',
  host='myredshiftcluster',
  port=5432,
  password='somepassword',
  async_=1,
  sslmode="require"
)
wait(conn)
acurs = conn.cursor()
acurs.execute('call public.test_sp(\'xyz\')')

【参考方案1】：

ODBC 和 JDBC 连接是同步的，因此构建异步进程将无法正常工作。幸运的是，AWS 最近宣布了 Redshift Data API，它是一个异步 REST 接口。因此，您可以通过该方法执行您正在寻找的内容。

见：https://docs.aws.amazon.com/redshift/latest/mgmt/data-api.html

【讨论】：

你是救生员。这非常适合我的需要。非常感谢。

很高兴听到这个消息。