创建表并写入数据

前提条件

• 已安装Aura DataFrame SDK。具体操作，请参见Aura DataFrame SDK获取与安装。
• 已创建AuraJobV2端点，并获取端点名称。具体操作，请参见创建Aura引擎端点。
• 已创建LakeFormation实例，并在该实例中创建Catalog及数据库。具体操作，请参见创建LakeFormation实例并规划元数据。
• 已创建并行文件系统。具体操作，请参见创建并行文件系统。
• 代码中涉及os.getenv()中的参数已配置到环境变量。
• 运行时环境要求Python版本为3.11。

示例

本示例创建三个数据表（text_samples、sales_line_items、category_sales）并分别写入文本、销售明细和分类销售数据，为后续章节（3.1.6.4.4，3.1.6.4.5，3.1.6.4.6，3.1.6.4.7，3.1.6.4.8，3.1.6.4.9，3.1.6.4.10）的查询、分析和UDF使用提供数据基础。

1. 导入依赖并连接AuraJobV2端点。
2. 创建text_samples表（包含id和content列），并插入5条文本数据。
3. 创建sales_line_items表（包含quantity、unit_price、discount_rate、tax_rate、return_flag、line_status、ship_date共7列），并插入5条销售数据。
4. 创建category_sales表（包含category和amount列），并插入12条分类销售数据。
5. 关闭AuraJobV2端点连接。

import os
import logging
import ibis
from aura_frame.multimodal import ai_lake

target_database = "test"  #在LakeFormation中创建数据库，例如：test。具体操作，请参见创建LakeFormation实例并规划元数据
#连接AuraJobV2端点，具体参数说明，请参见建立与AuraJobV2端点的连接。
con = ai_lake.connect(
    aura_endpoint=os.getenv("aura_endpoint"),         # AuraJobV2端点所在的终端节点（Endpoint）
    aura_endpoint_id=os.getenv("aura_endpoint_id"),    # AuraJobV2端点ID
    aura_endpoint_name=os.getenv("aura_endpoint_name"),# AuraJobV2端点名称
    aura_workspace_id=os.getenv("aura_workspace_id"),  # 工作空间ID
    lf_catalog_name=os.getenv("lf_catalog_name"),      # LakeFormation Catalog名称
    access_key=os.getenv("access_key"),                # 访问密钥ID（AK）
    secret_key=os.getenv("secret_key"),                # 访问密钥（SK）
    default_database=target_database,                       # LakeFormation数据库名称
    use_single_cn_mode=True,   # 是否开启单CN模式
    logging_level=logging.INFO # 设置日志级别
)
try:
    table_name = "text_samples"
    table_schema = ibis.schema({"id": "int32", "content": "string"})
    con.sql(f'DROP TABLE IF EXISTS "{target_database}"."{table_name}";')
    con.create_table(
        table_name,
        schema=table_schema,
        database=target_database,
        table_format="orc",
        if_not_exists=True,
    )
    insert_sql = f'''
    INSERT INTO "{target_database}"."{table_name}" (id, content) VALUES
        (1, 'Hello, how are you today?'),
        (2, 'The quick brown fox jumps over the lazy cat.'),
        (3, 'Machine learning is transforming the way we process data.'),
        (4, 'Natural language processing enables computers to understand human language.'),
        (5, 'Tokenization is the first step in text processing pipelines.')
    '''
    con.sql(insert_sql)
    print(f"表 {table_name} 创建并写入数据完成")
    table_name = "sales_line_items"
    table_schema = ibis.schema({
        "quantity": "int32",
        "unit_price": "float64",
        "discount_rate": "float64",
        "tax_rate": "float64",
        "return_flag": "string",
        "line_status": "string",
        "ship_date": "date",
    })
    con.sql(f'DROP TABLE IF EXISTS "{target_database}"."{table_name}";')
    con.create_table(
        table_name,
        schema=table_schema,
        database=target_database,
        table_format="orc",
        if_not_exists=True,
    )
    insert_sql = f'''
    INSERT INTO "{target_database}"."{table_name}" (quantity, unit_price, discount_rate, tax_rate, return_flag, line_status, ship_date) VALUES
        (10, 1000.00, 0.05, 0.08, 'N', 'O', '1998-08-01'),
        (20, 2000.00, 0.10, 0.05, 'R', 'F', '1998-09-15'),
        (15, 1500.00, 0.08, 0.07, 'N', 'O', '1998-07-20'),
        (25, 2500.00, 0.05, 0.08, 'R', 'F', '1998-10-01'),
        (30, 3000.00, 0.02, 0.10, 'N', 'O', '1998-08-15')
    '''
    con.sql(insert_sql)
    print(f"表 {table_name} 创建并写入数据完成")
    table_name = "category_sales"
    table_schema = ibis.schema({"category": "string", "amount": "float64"})
    con.sql(f'DROP TABLE IF EXISTS "{target_database}"."{table_name}";')
    con.create_table(
        table_name,
        schema=table_schema,
        database=target_database,
        table_format="orc",
        if_not_exists=True,
    )
    insert_sql = f'''
    INSERT INTO "{target_database}"."{table_name}" (category, amount) VALUES
        ('Electronics', 1299.00),
        ('Electronics', 799.00),
        ('Electronics', 599.00),
        ('Electronics', 1099.00),
        ('Books', 29.99),
        ('Books', 49.99),
        ('Books', 19.99),
        ('Books', 39.99),
        ('Clothing', 89.00),
        ('Clothing', 159.00),
        ('Clothing', 49.00),
        ('Clothing', 199.00)
    '''
    con.sql(insert_sql)
    print(f"表 {table_name} 创建并写入数据完成")
finally:
    con.close()

相关文档

• Session APIs
• Dataset APIs
• Table APIs