将 Snowflake SQLAlchemy 工具包与 Python Connector 结合使用¶

Snowflake Connector for Python¶

Snowflake SQLAlchemy 的唯一要求是 Snowflake Connector for Python；但是，不需要安装该连接器，因为安装 Snowflake SQLAlchemy 会自动安装连接器。

数据分析和 Web 应用程序框架（可选）¶

Snowflake SQLAlchemy 可以与 pandas、Jupyter 和 Pyramid 一起使用，它们为数据分析和 Web 应用程序提供了更高级别的应用程序框架。但是，从零开始构建工作环境并不是一项简单的任务，特别是对于新手用户来说。安装框架需要 C 编译器和工具，选择正确的工具和版本是一个障碍，可能会阻止用户使用 Python 应用程序。

构建环境的一种更简单的方法是通过 Anaconda，它为所有用户（包括数据分析师和学生等非 Python 专家）提供了完整的预编译技术堆栈。有关 Anaconda 安装说明，请参阅 Anaconda 安装文档。然后可以使用 pip 将 Snowflake SQLAlchemy 包安装在 Anaconda 之上。

连接参数¶

'snowflake://<user_login_name>:<password>@<account_identifier>'

'snowflake://<user_login_name>:<password>@<account_identifier>/<database_name>/<schema_name>?warehouse=<warehouse_name>&role=<role_name>'

打开和关闭连接¶

通过执行 engine.connect() 打开连接；避免使用 engine.execute()。

# Avoid this.
engine = create_engine(...)
engine.execute(<SQL>)
engine.dispose()

# Do this.
engine = create_engine(...)
connection = engine.connect()
try:
    connection.execute(<SQL>)
finally:
    connection.close()
    engine.dispose()

Copy

如果计划使用显式事务，则必须禁用 SQLAlchemy 中的 AUTOCOMMIT 执行选项。

默认情况下，SQLAlchemy 启用此选项。启用此选项后，INSERT、UPDATE 和 DELETE 语句在执行时会自动提交，即使这些语句在显式事务中运行也是如此。

若要禁用 AUTOCOMMIT，请将 autocommit=False 传递给 Connection.execution_options() 方法。例如：

# Disable AUTOCOMMIT if you need to use an explicit transaction.
with engine.connect().execution_options(autocommit=False) as connection:

  try:
    connection.execute("BEGIN")
    connection.execute("INSERT INTO test_table VALUES (88888, 'X', 434354)")
    connection.execute("INSERT INTO test_table VALUES (99999, 'Y', 453654654)")
    connection.execute("COMMIT")
  except Exception as e:
    connection.execute("ROLLBACK")
  finally:
    connection.close()

engine.dispose()

自动递增行为¶

自动递增一个值需要 Sequence 对象。将 Sequence 对象包含在主键列中，以便在插入每条新记录时自动递增此值。例如：

t = Table('mytable', metadata,
    Column('id', Integer, Sequence('id_seq'), primary_key=True),
    Column(...), ...
)

Copy

对象名称大小写处理¶

Snowflake 将所有不区分大小写的对象名称存储在大写文本中。相反，SQLAlchemy 将所有小写对象名称视为不区分大小写。Snowflake SQLAlchemy 在架构级通信期间（即在表和索引反射期间）转换对象名称大小写。如果使用大写对象名称，SQLAlchemy 则假定它们区分大小写，并将名称放在引号内。此行为将导致与从 Snowflake 接收的数据字典数据不匹配，因此除非标识符名称真正使用引号创建为区分大小写（例如 "TestDb"），否则应在 SQLAlchemy 端使用所有小写名称。

索引支持¶

索引仅在 Snowflake SqlAlchemy 的混合表中受支持。有关限制和用例的更多详细信息，请参阅 创建索引 文档。您可以使用以下方法创建索引：

• 单列索引

  您可以通过在列上设置 index=True 参数或通过明确定义 Index 对象来创建单列索引。

  hybrid_test_table_1 = HybridTable(
    "table_name",
    metadata,
    Column("column1", Integer, primary_key=True),
    Column("column2", String, index=True),
    Index("index_1","column1", "column2")
  )
  
  metadata.create_all(engine_testaccount)
  

  Copy

• 多列索引

  对于多列索引，您需要定义索引对象，指定应被索引的列。

  hybrid_test_table_1 = HybridTable(
    "table_name",
    metadata,
    Column("column1", Integer, primary_key=True),
    Column("column2", String),
    Index("index_1","column1", "column2")
  )
  
  metadata.create_all(engine_testaccount)
  

  Copy

Numpy 数据类型支持¶

Snowflake SQLAlchemy 支持绑定和提取 NumPy 数据类型。始终支持绑定。若要启用提取 NumPy 数据类型，请将 numpy=True 添加到连接参数。

支持以下 NumPy 数据类型：

• numpy.int64

• numpy.float64

• numpy.datetime64

以下示例显示了 numpy.datetime64 数据的往返行程：

import numpy as np
import pandas as pd
engine = create_engine(URL(
    account = 'myorganization-myaccount',
    user = 'testuser1',
    password = 'pass',
    database = 'db',
    schema = 'public',
    warehouse = 'testwh',
    role='myrole',
    numpy=True,
))

specific_date = np.datetime64('2016-03-04T12:03:05.123456789Z')

with engine.connect() as connection:
    connection.exec_sql_query(
        "CREATE OR REPLACE TABLE ts_tbl(c1 TIMESTAMP_NTZ)")
    connection.exec_sql_query(
        "INSERT INTO ts_tbl(c1) values(%s)", (specific_date,)
    )
    df = pd.read_sql_query("SELECT * FROM ts_tbl", connection)
    assert df.c1.values[0] == specific_date

Copy

缓存列元数据¶

SQLAlchemy 提供运行时检查 API，以获取有关各种对象的运行时信息。一个常见的用例是，获取架构中的所有表及其列元数据，以便构建架构目录。

伪代码流程如下：

inspector = inspect(engine)
schema = inspector.default_schema_name
for table_name in inspector.get_table_names(schema):
    column_metadata = inspector.get_columns(table_name, schema)
    primary_keys = inspector.get_primary_keys(table_name, schema)
    foreign_keys = inspector.get_foreign_keys(table_name, schema)
    ...

Copy

在此流程中，一个潜在的问题是，在每个表上运行查询时可能需要相当长的时间。结果是缓存的，但获取列元数据的成本很高。

为了缓解此问题，Snowflake SQLAlchemy 采用一个标志 cache_column_metadata=True，以便在调用 get_table_names 时缓存所有表的所有列元数据，且 get_columns 的其余部分、get_primary_keys 和 get_foreign_keys 也可以利用该缓存。

engine = create_engine(URL(
    account = 'myorganization-myaccount',
    user = 'testuser1',
    password = 'pass',
    database = 'db',
    schema = 'public',
    warehouse = 'testwh',
    role='myrole',
    cache_column_metadata=True,
))

Copy

VARIANT、ARRAY 和 OBJECT 支持¶

Snowflake SQLAlchemy 支持提取 VARIANT、ARRAY 和 OBJECT 数据类型。所有类型都将在 Python 中转换为 str，以便您可以使用 json.loads 将它们转换为原生数据类型。

此示例展示了如何创建包含 VARIANT、ARRAY 和 OBJECT 数据类型列的表：

from snowflake.sqlalchemy import (VARIANT, ARRAY, OBJECT)
...
t = Table('my_semi_structured_datatype_table', metadata,
    Column('va', VARIANT),
    Column('ob', OBJECT),
    Column('ar', ARRAY))
metadata.create_all(engine)

Copy

为了检索 VARIANT、 ARRAY 和 OBJECT 数据类型列并将它们转换为原生 Python 数据类型，请提取数据并调用 json.loads 方法，如下所示：

import json
connection = engine.connect()
results = connection.execute(select([t]))
row = results.fetchone()
data_variant = json.loads(row[0])
data_object  = json.loads(row[1])
data_array   = json.loads(row[2])

Copy

结构化数据类型支持¶

本模块为 Snowflake 结构化数据（特别是 Iceberg 表）定义了自定义 SQLAlchemy 类型。MAP、OBJECT 和 ARRAY 类型允许在 SQLAlchemy 模型中存储复杂的数据结构。有关详细信息，请参阅 Snowflake 结构化数据类型 文档。

IcebergTable(
  table_name,
  metadata,
  Column("id", Integer, primary_key=True),
  Column("map_col", MAP(NUMBER(10, 0), TEXT(16777216))),
  external_volume="external_volume",
  base_location="base_location",
)

IcebergTable(
    table_name,
    metadata,
    Column("id", Integer, primary_key=True),
    Column(
        "object_col",
        OBJECT(key1=(TEXT(16777216), False), key2=(NUMBER(10, 0), False)),
        OBJECT(key1=TEXT(16777216), key2=NUMBER(10, 0)), # Without nullable flag
    ),
    external_volume="external_volume",
    base_location="base_location",
)

IcebergTable(
    table_name,
    metadata,
    Column("id", Integer, primary_key=True),
    Column("array_col", ARRAY(TEXT(16777216))),
    external_volume="external_volume",
    base_location="base_location",
)

CLUSTER BY 支持¶

Snowflake SQLAlchemy 支持表的 CLUSTER BY 参数。有关该参数的信息，请参阅 CREATE TABLE。

此示例展示了如何将包含两列（id 和 name）的表创建为群集密钥：

t = Table('myuser', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
    snowflake_clusterby=['id', 'name'], ...
)
metadata.create_all(engine)

Copy

Alembic 支持¶

Alembic 是 SQLAlchemy 之上的数据库迁移工具。Snowflake SQLAlchemy 的运作方式是将以下代码添加到 alembic/env.py，以便 Alembic 可以识别 Snowflake SQLAlchemy。

from alembic.ddl.impl import DefaultImpl

class SnowflakeImpl(DefaultImpl):
    __dialect__ = 'snowflake'

Copy

密钥对身份验证支持¶

Snowflake SQLAlchemy 通过利用 Snowflake Connector for Python 函数来支持密钥对身份验证。有关创建私钥和公钥的步骤，请参阅 使用密钥对身份验证和密钥对轮换。

私钥参数通过 connect_args 来传递，如下所示：

...
from snowflake.sqlalchemy import URL
from sqlalchemy import create_engine

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization

with open("rsa_key.p8", "rb") as key:
    p_key= serialization.load_pem_private_key(
        key.read(),
        password=os.environ['PRIVATE_KEY_PASSPHRASE'].encode(),
        backend=default_backend()
    )

pkb = p_key.private_bytes(
    encoding=serialization.Encoding.DER,
    format=serialization.PrivateFormat.PKCS8,
    encryption_algorithm=serialization.NoEncryption())

engine = create_engine(URL(
    account='abc123',
    user='testuser1',
    ),
    connect_args={
        'private_key': pkb,
        },
    )

Copy

其中 PRIVATE_KEY_PASSPHRASE 是用于解密私钥文件 (rsa_key.p8) 的密码。

snowflake.sqlalchemy.URL 方法不支持私钥参数。

合并命令支持¶

Snowflake SQLAlchemy 支持使用其 MergeInto 自定义表达式执行更新或插入。有关完整文档，请参阅 MERGE。

按如下方式使用：

from sqlalchemy.orm import sessionmaker
from sqlalchemy import MetaData, create_engine
from snowflake.sqlalchemy import MergeInto

engine = create_engine(db.url, echo=False)
session = sessionmaker(bind=engine)()
connection = engine.connect()

meta = MetaData()
meta.reflect(bind=session.bind)
t1 = meta.tables['t1']
t2 = meta.tables['t2']

merge = MergeInto(target=t1, source=t2, on=t1.c.t1key == t2.c.t2key)
merge.when_matched_then_delete().where(t2.c.marked == 1)
merge.when_matched_then_update().where(t2.c.isnewstatus == 1).values(val = t2.c.newval, status=t2.c.newstatus)
merge.when_matched_then_update().values(val=t2.c.newval)
merge.when_not_matched_then_insert().values(val=t2.c.newval, status=t2.c.newstatus)
connection.execute(merge)

Copy

CopyIntoStorage 支持¶

Snowflake SQLAlchemy 支持使用其自定义 CopyIntoStorage 表达式将表和查询结果保存到不同的 Snowflake 暂存区、Azure 容器和 AWS 桶中。有关完整文档，请参阅 COPY INTO <location>。

按如下方式使用：

from sqlalchemy.orm import sessionmaker
from sqlalchemy import MetaData, create_engine
from snowflake.sqlalchemy import CopyIntoStorage, AWSBucket, CSVFormatter

engine = create_engine(db.url, echo=False)
session = sessionmaker(bind=engine)()
connection = engine.connect()

meta = MetaData()
meta.reflect(bind=session.bind)
users = meta.tables['users']

copy_into = CopyIntoStorage(from_=users,
                            into=AWSBucket.from_uri('s3://my_private_backup').encryption_aws_sse_kms('1234abcd-12ab-34cd-56ef-1234567890ab'),
                            formatter=CSVFormatter().null_if(['null', 'Null']))
connection.execute(copy_into)

Copy

Iceberg 表与 Snowflake 目录支持¶

Snowflake SQLAlchemy 支持带有 Snowflake 目录的 Iceberg 表，以及各种相关参数。有关 Iceberg 表的详细信息，请参阅 Snowflake CREATE ICEBERG 文档。

若要使用 Snowflake SQLAlchemy 创建 Iceberg 表，您可以使用 SQLAlchemy Core 语法定义表，如下所示：

table = IcebergTable(
        "myuser",
        metadata,
        Column("id", Integer, primary_key=True),
        Column("name", String),
        external_volume=external_volume_name,
        base_location="my_iceberg_table",
  as_query="SELECT * FROM table"
    )

或者，您可以使用声明式方法定义表：

class MyUser(Base):
    __tablename__ = "myuser"

    @classmethod
    def __table_cls__(cls, name, metadata, *arg, **kw):
        return IcebergTable(name, metadata, *arg, **kw)

    __table_args__ = {
        "external_volume": "my_external_volume",
        "base_location": "my_iceberg_table",
  "as_query": "SELECT * FROM table",
    }

    id = Column(Integer, primary_key=True)
    name = Column(String)

混合表支持¶

Snowflake SQLAlchemy 支持具有索引的混合表。有关详细信息，请参阅 Snowflake CREATE HYBRID TABLE 文档。

若要创建混合表并添加索引，您可以使用 SQLAlchemy 核心语法，如下所示：

table = HybridTable(
    "myuser",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Index("idx_name", "name")

此外，您也可以使用声明式方法定义表：

class MyUser(Base):
    __tablename__ = "myuser"

    @classmethod
    def __table_cls__(cls, name, metadata, *arg, **kw):
        return HybridTable(name, metadata, *arg, **kw)

    __table_args__ = (
        Index("idx_name", "name"),
    )

    id = Column(Integer, primary_key=True)
    name = Column(String)

动态表支持¶

Snowflake SQLAlchemy 支持动态表。有关详细信息，请参阅 Snowflake CREATE DYNAMIC TABLE 文档。

若要创建动态表，您可以使用 SQLAlchemy 核心语法，如下所示：

 dynamic_test_table_1 = DynamicTable(
       "dynamic_MyUser",
       metadata,
       Column("id", Integer),
       Column("name", String),
       target_lag=(1, TimeUnit.HOURS), # Additionally you can use SnowflakeKeyword.DOWNSTREAM
       warehouse='test_wh',
refresh_mode=SnowflakeKeyword.FULL
       as_query="SELECT id, name from MyUser;"
   )

此外，您也可以使用 SqlAlchemy select() 结构定义一个没有列的表：

 dynamic_test_table_1 = DynamicTable(
       "dynamic_MyUser",
       metadata,
       target_lag=(1, TimeUnit.HOURS),
       warehouse='test_wh',
refresh_mode=SnowflakeKeyword.FULL
       as_query=select(MyUser.id, MyUser.name)
   )