[Snowflake+Python]Python中如何使用Snowflake

Snowflake + Python

使用场景：自动化/构建数据管道以及在预处理后将数据存储到Snowflake中.

Snowflake Setup:

11.登录： https ://www.snowflake.com/

输入详细信息并申请 30 天试用，通过 Gmail 验证。
在试用中，Snowflake 将提供足够的积分来开始使用。

2.下载雪花 CLI：

Redirecting to … https://docs.snowflake.net/manuals/user-guide/snowsql-install-config.html

注意：Snowflake 的所有服务都有很好的文档。查看此用户指南以获得更详细的理解 (Redirecting to … https://docs.snowflake.net/manuals/user-guide.html

与雪花互动

Snowflake 有 2 种方式可以与服务交互：

Web 界面： Snowflake 提供了一个 Web 界面，其中包括在 Snowflake 中执行创建、修改和管理帐户和资源等操作的工具。

Web 界面
的限制：“put”命令不能通过 Web 界面运行。
湾。无法加载大于 50MB 的文件，因为它旨在用于小文件。

CLI：如果您是 Linux 用户，则该界面易于使用，并填补了 Web 界面的这些空白。

为表演搭建舞台（如果您熟悉 Snowflake 的设计流程，请跳过）

雪花的架构

Snowflake 的独特架构由三个关键层组成：
1. 集中式存储
2. 多集群计算和处理
3. 云服务

集中存储

当数据加载到 Snowflake 中时，它会将数据重新组织为 Snowflake 内部优化、压缩的列格式。Snowflake 将这些优化的数据存储在云存储中。Snowflake 管理如何存储这些数据的所有方面。

查询处理

每个虚拟仓库都是一个独立的计算集群，不与其他虚拟仓库共享计算资源。因此，每个虚拟仓库对其他虚拟仓库的性能没有影响。

云服务

该层内的服务包括：

• 验证
• 基础设施管理
• 元数据管理
• 查询解析和优化
• 访问控制

使用不同的权限，管理不同的数据库。因此，用户的角色对于执行某些任务是必不可少的。

设置 Python 和代码依赖项

参考：https ://docs.snowflake.net/manuals/user-guide/python-connector-install.html

确保安装了 Python 3.x 和所需的模块。

为 Snowflake 安装 Python 包：

pip install — upgrade snowflake-connector-python

py 脚本以非常简单的方式编写，没有花哨的类或方法。

代码（git repo）：hashmapinc / oss / python_snowflake_poc · GitLab
（这包括 Jupyter notebook）

注意：repo 还包含 Docker 映像和文件。暂时不要担心，因为这不是我们本次练习的重点。

1. 导入包

“cred.json” -> 包含用户信息的 JSON 文件，以避免在代码中写入敏感信息。我将 JSON 加载到 dict 类型的“cred”变量中，如下所示：

# import require module and credential
import snowflake.connector
import json
with open(“cred.json”,”r”) as f:
cred = json.load(f)
创建“cred.json” JSON文件并写入或者你可以使用用于创建 JSON 的 Json 转储：
{
“userid”:”userid”,
“password”:”xxxxx”,
“account”:”px00000.ap-southeast-2”
}

2. 通过 Python 设置与 Snowflake 的连接

cOnn= snowflake.connector.connect(
user=cred["userid"],
password=cred["password"],
account=cred["account"],
session_parameters={
"QUERY_TAG": "EndOfMonthFinance",
}
)
print( “连接成功”，conn）

您还可以通过在连接后执行 SQL 语句 ALTER SESSION SET ... 来设置会话参数：

con.cursor().execute("ALTER SESSION SET QUERY_TAG = 'EndOfMonthFinancials'")
"""

3. 将当前角色更改为“ SYSADMIN”，因为默认可能不是 SYSADMIN。这对于执行将要执行的操作很重要。

conn.cursor().execute("使用角色 sysadmin")

4.设置雪花存储和管理数据：

使用“ CREATE OR REPLACE” -> 用于新建和覆盖
使用“CREATE” -> 用于新建（如果存在则创建时出错）
或使用IF NOT EXISTS

# 创建数据库、模式和仓库
conn.cursor().execute("CREATE WAREHOUSE IF NOT EXISTS tiny_warehouse_mg")
conn.cursor().execute("CREATE DATABASE IF NOT EXISTS testdb_mg")
conn.cursor().execute ("USE DATABASE testdb_mg")
# 上面的行必须在下面的行之前执行，因为在构建 WH 的 testschema_mg 基本结构时，DB 应该存在
conn.cursor().execute("CREATE SCHEMA IF NOT EXISTS testschema_mg")

5.设置当前需求WH、DB、Schema。

# 使用数据库、模式和仓库
conn.cursor().execute("USE DATABASE testdb_mg")
conn.cursor().execute("USE WAREHOUSE tiny_warehouse_mg")
conn.cursor().execute("USE DATABASE testdb_mg" )
conn.cursor().execute("使用 SCHEMA testdb_mg.testschema_mg")

6.创建一个2列的新表；一个 int 类型和另一个 string 类型

# 创建表并插入数据
conn.cursor().execute(
"CREATE OR REPLACE TABLE "
"test_table(col1 integer, col2 string)" )

7.让我们在上表中插入2条记录

这里我们使用字符串连接。与 Python 中的任何其他 DB 连接器一样，我们也可以使用元组来防止SQL 注入

# Bad query:
conn.cursor().execute(
"INSERT INTO test_table(col1, col2) VALUES " +
" (123, 'indian Cricket'), " +
" (100, 'Kapil Dev')")
安全方法：
# 安全示例。做这个！
conn.cursor().execute("INSERT INTO test_table(col1, col2) VALUES (%s, %s)", ('123', 'indian Cricket'));

8. 将 Python 与 Snowflake 一起使用

您必须先在雪花数据存储中加载数据，然后才能对该数据运行查询。为了正确管理它，我们有一个暂存区域和模式类型来帮助我们以所需的严格格式提取数据。下面的代码将帮助我们加载目录名称“data”中存在的 CSV 文件。在暂存它们之后，我们将复制文件内容或将数据转储到我们创建的表（test_table）中。

如果您使用的是Window 的 cmd，则必须提供类似（“\\”）的路径
：（ “PUT file://.\\data\\crick* @testdb_mg.testschema_mg.%test_table”）

对于 S3 或任何其他云存储，需要类似的代码。

注意：'put' 是从 CLI 中使用的，并确保有访问权限，提到的路径。

# Putting Data
# @DB .SCHEMA.TABLE ，这是根据内容的层次结构。like DB <- SCHEMA <- Table
# 如果使用 Window 的 cmd : ("PUT file://.\\data\\crick* @testdb_mg .testschema_mg.%test_table")
conn.cursor().execute("PUT file: //./data/crick* @testdb_mg .testschema_mg.%test_table")
conn.cursor().execute("""COPY INTO test_table from @testdb_mg .testschema_mg.%test_table/crick*.csv.gz
file_format = (type = csv field_delimiter=',')
模式 = '.*.csv.gz'
on_error= 'skip_file'""")# 对于 S3
# 复制数据
# con.cursor().execute("""
# COPY INTO testtable FROM s3:///data/
# CREDENTIALS = (
# aws_key_id='{aws_access_key_id}',
# aws_secret_key=' {aws_secret_access_key}')
# FILE_FORMAT=(field_delimiter=',')
# """.format(
#aws_access_key_id=AWS_ACCESS_KEY_ID,
#aws_secret_access_key=AWS_SECRET_ACCESS_KEY))

9. 我们现在在表格中有数据

# 查询数据
cur = conn.cursor()
try:
cur.execute("SELECT col1, col2 FROM test_table ORDER BY col1")
for (col1, col2) in cur:
print('{0}, {1}'.format (col1, col2))
finally:
cur.close() # 为系统稳定性避免不必要的连接
print("connection closed, script end.. Thanks ")
# 如果结果集太大而无法放入内存，则使用 fetchone 或 fetchmany。
# 结果 = conn.cursor().execute("SELECT col1, col2 FROM test_table").fetchall()

转载： https://medium.com/hashmapinc/getting-started-with-snowflake-using-python-8ec552ffd02