在雪花中从外部阶段动态创建表

Question

1）Python到雪花数据的连接是使用python连接器进行的 2）外部阶段设置指向S3桶 要求是根据每个 CSV 创建一个动态表。

例如-我的 s3 存储桶中有 10 个 CSV，那么应该参考外部阶段动态创建 10 个不同的表

Sql_query=？

Sql_query=从@db.schema.external_stage复制到db.schema.table（在雪花中已经创建表结构的情况下）

Answer 1

请参阅Python - 雪花连接。

参考使用python从雪花中查询数据。

下面的示例将读取包含两个文件的阶段并 根据阶段中的文件名创建两个表并 然后将数据插入到每个表中。

这就是舞台的样子[输出被截断，按列]-

list @test_stage;
+------------------------------+------+
| name                         | size |
|------------------------------+------+
| test_stage/data_comma.csv.gz |  128 |
| test_stage/date_data.csv.gz  |  128 |
+------------------------------+------+

用于创建动态表并从舞台插入数据的Python代码

import snowflake.connector

# Below setting up connection properties
# Providing all values for database, schema
# therefore prefixing those values later in code
# is not needed - can be modified as needed.

con = snowflake.connector.connect(
    user='user',
    password='password',
    account='ab313.us-east-2.aws',
    warehouse='COMPUTE_WH',
    database='TEST_DB',
    schema='PUBLIC'
)

try:
    cur = con.cursor()
    
    # Query stage and get file-names
    # Split file name to extract part NOT containing extensions
    
    cur.execute("select DISTINCT split_part(metadata$filename,'.',1) as col1, metadata$filename as col2 from @test_stage;")
    for (col1,col2) in cur:
        # Dynamic table creation is based on files in stage.
        # Here stage files are with skip-header, so column name are 
        # hard-coded.
        # Even if we try and get column-names from header row 
        # (if included) in stage file, data-types still needs 
        # hard-coding or perhaps fetching from other table 
        # if needed complete dynamism.
        cur.execute("create transient table "+col1+"(id number);")
        
        # Copy data into table created above from stage file
        cur.execute("copy into "+col1+" from (select $1 from @test_stage/"+col2+");")    
finally:
    con.close()

Answer 2

观看此播放列表：https://youtu.be/GfhNaceyzDw?feature=shared .

这应该可以加深您的理解和解决方案。