如何使用 boto3 或 psycopg2 python 库在 redshift 中插入数据

Question

在 python lambda 函数中，最适合使用“boto3”和“Psycopg2”进行 redshift 操作的库：

• 在 redshift 集群中查找表 在 redshift
• 集群中创建表
• 在 redshift 集群中插入数据

如果我得到以下答复，我将不胜感激：

• 满足上述所有 3 个需求的任一库的 python 代码。

提前致谢！！

Answer 1

使用 psycopg2 从 Lambda 直接连接到 Redshift 是更简单、更直接的方法，但存在很大的限制。 Lambda 函数有运行时间限制，即使您的 SQL 命令没有超过最大运行时间，您也需要为 Lambda 函数付费以等待 Redshift 完成 SQL。对于快速运行的 SQL 命令，运行速度很快，这不是问题，但插入数据可能需要一些时间，具体取决于数据量。

如果您所有的 Redshift 操作都不到几秒（并且不会随着时间的推移而变得更长），那么 psycopg2 直接连接到 Redshift 可能是最佳选择。如果数据插入需要一两分钟，但这个过程不经常运行（每天），那么 psycopg2 可能仍然是最佳选择，因为频繁运行时 Lambda 并不是很昂贵。这是流程简单性与成本计算的比较。

使用 Redshift Data API 更为复杂。此过程允许您将 SQL 触发到 Redshift 并终止 Lambda。稍后运行的 Lambda 将检查 SQL 是否已完成，并检查 SQL 的结果。 SQL 未完成意味着需要稍后调用 Lambda 以查看事情是否完成。此轮询过程通常由 Step Function 和一组不同的 Lambda 函数完成。不是超级困难，但复杂程度高于单个 Lambda。由于这是一个轮询过程，因此在检查结果之间存在等待时间，如果太长会导致延迟，如果太短则会导致过度轮询和额外成本。

如果您因超时原因需要使用 Data API，那么您可能需要同时使用 psycopg2 来对数据库进行短期运行查询 - 例如“此表是否存在？”。使用数据 API 执行长时间运行的步骤，例如“将这组 1TB 数据插入 Redshift”。

Answer 2

使用 boto3 执行所有三个操作的示例基本 Python 代码。

import json
import boto3

clientdata = boto3.client('redshift-data')

# looks up table and returns true if found
def lookup_table(table_name):
  response = clientdata.list_tables(
    ClusterIdentifier='redshift-cluster-1',
    Database='dev',
    DbUser='awsuser',
    TablePattern=table_name
  )
  print(response)
  if ( len(response['Tables']) == 0 ):
    return False
  else:
    return True

# creates table with one integer column
def create_table(table_name):
  sqlstmt = 'CREATE TABLE '+table_name+' (col1 integer);'
  print(sqlstmt)
  response = clientdata.execute_statement(
    ClusterIdentifier='redshift-cluster-1',
    Database='dev',
    DbUser='awsuser',
    Sql=sqlstmt,
    StatementName='CreateTable'
  )
  print(response)

# inserts one row with integer value for col1
def insert_data(table_name, dval):
  print(dval)
  sqlstmt = 'INSERT INTO '+table_name+'(col1) VALUES ('+str(dval)+');'
  response = clientdata.execute_statement(
    ClusterIdentifier='redshift-cluster-1',
    Database='dev',
    DbUser='awsuser',
    Sql=sqlstmt,
    StatementName='InsertData'
  )
  print(response)

result = lookup_table('date')
if ( result ):
  print("Table exists.")
else:
  print("Table does not exist!")

create_table("testtab")
insert_data("testtab", 11)

我没有使用 Lambda，而是仅从我的 shell 执行它。
希望这有帮助。假设已经为客户端设置了凭据和默认区域。