使用 Amazon Redshift SQLAlchemy 方言与 Amazon Redshift 交互

pip install sqlalchemy-redshift
pip install redshift_connector

import sqlalchemy as sa
from sqlalchemy.engine.url import URL
from sqlalchemy import orm as sa_orm from sqlalchemy_redshift.dialect import TIMESTAMPTZ, TIMETZ # build the sqlalchemy URL. When authenticating using IAM, the host
# and port do not need to be specified by the user.
url = URL.create(
drivername='redshift+redshift_connector', # indicate redshift_connector driver and dialect will be used
database='dev', # Amazon Redshift database
username='johnd@example.com', # Okta username
password='<PWD>' # Okta password
) # a dictionary is used to store additional connection parameters
# that are specific to redshift_connector or cannot be URL encoded.
conn_params = { "iam": True, # must be enabled when authenticating via IAM "credentials_provider": "OktaCredentialsProvider", "idp_host": "<prefix>.okta.com", "app_id": "<appid>", "app_name": "amazon_aws_redshift", "region": "<region>", "cluster_identifier": "<clusterid>", "ssl_insecure": False, # ensures certificate verification occurs for idp_host
} engine = sa.create_engine(url, connect_args=conn_params)

import sqlalchemy as sa
from sqlalchemy.engine.url import URL # build the sqlalchemy URL
url = URL.create(
drivername='redshift+redshift_connector', # indicate redshift_connector driver and dialect will be used
host='<clusterid>.xxxxxx.<aws-region>.redshift.amazonaws.com', # Amazon Redshift host
port=5439, # Amazon Redshift port
database='dev', # Amazon Redshift database
username='awsuser', # Amazon Redshift username
password='<pwd>' # Amazon Redshift password
) engine = sa.create_engine(url) Next, we will create a session using the already established engine above. Session = sa_orm.sessionmaker()
Session.configure(bind=engine)
session = Session() # Define Session-based Metadata
metadata = sa.MetaData(bind=session.bind)

import datetime
import uuid
import random table_name = 'product_clickstream_tz' RedshiftDBTable = sa.Table(
table_name,
metadata,
sa.Column('session_id', sa.VARCHAR(80)),
sa.Column('click_region', sa.VARCHAR(100), redshift_encode='lzo'),
sa.Column('product_id', sa.BIGINT),
sa.Column('click_datetime', TIMESTAMPTZ),
sa.Column('stream_time', TIMETZ),
sa.Column ('order_detail', SUPER),
redshift_diststyle='KEY',
redshift_distkey='session_id',
redshift_sortkey='click_datetime'
) # Drop the table if it already exists
if sa.inspect(engine).has_table(table_name):
RedshiftDBTable.drop(bind=engine) # Create the table (execute the "CREATE TABLE" SQL statement for "product_clickstream_tz")
RedshiftDBTable.create(bind=engine) In this step, you will populate the table by preparing the insert command. # create sample data set
# generate a UUID for this row
session_id = str(uuid.uuid1()) # create Region information
click_region = "US / New York" # create Product information
product_id = random.randint(1,100000) # create a datetime object with timezone
click_datetime = datetime.datetime(year=2021, month=10, day=20, hour=10, minute=12, second=40, tzinfo=datetime.timezone.utc) # create a time object with timezone
stream_time = datetime.time(hour=10, minute=14, second=56, tzinfo=datetime.timezone.utc) # create SUPER information
order_detail = '[{"o_orderstatus":"F","o_clerk":"Clerk#0000001991","o_lineitems":[{"l_returnflag":"R","l_tax":0.03,"l_quantity":4,"l_linestatus":"F"}]}]' # create the insert SQL statement
insert_data_row = RedshiftDBTable.insert().values(
session_id=session_id,
click_region=click_region,
product_id=product_id,
click_datetime=click_datetime,
stream_time=stream_time,
order_detail=order_detail
) # execute the insert SQL statement
session.execute(insert_data_row)
session.commit()

pip install ipython-sql

import sqlalchemy as sa
from sqlalchemy.engine.url import URL
from sqlalchemy.orm import Session
%reload_ext sql
%config SqlMagic.displaylimit = 25 connect_to_db = URL.create(
drivername='redshift+redshift_connector', host='cluster.xxxxxxxx.region.redshift.amazonaws.com', port=5439, database='dev', username='awsuser', password='xxxxxx' )
%sql $connect_to_db
%sql select current_user, version();

session.close() # If the connection was accessed directly, ensure it is invalidated
conn = engine.connect()
conn.invalidate() # Clean up the engine
engine.dispose()