如何用Python实现SAP HANA多表导出至BigQuery并批量建表

要实现多表批量导出，核心思路是先获取需要同步的HANA表列表，再遍历每个表完成数据读取与BigQuery写入，以下是具体实现步骤和优化后的代码：

关键步骤说明 #

1. 获取HANA待导出表列表：通过查询HANA系统视图SYS.TABLES获取指定模式下的所有用户表（过滤系统表）。
2. 遍历处理每个表：对每个表执行查询、数据转换、BigSchema生成、写入操作。
3. 自动生成BigQuery Schema：利用Pandas DataFrame的结构自动推断Schema，避免手动逐个定义。
4. 表名映射：将HANA表名直接映射为BigQuery的目标表名（可根据需求调整命名规则）。

优化后的完整代码 #

from hdbcli import dbapi
import pandas as pd
from google.cloud import bigquery
from google.cloud.bigquery.schema import SchemaField
from google.cloud.bigquery.enums import SqlTypeNames

# 初始化BigQuery客户端
client = bigquery.Client()

# 连接SAP HANA数据库
conn = dbapi.connect(
    address="*******.hana.trial-***10.hanacloud.ondemand.com", 
    port=443,
    user='DBADMIN', 
    password='******'
)

if not conn.isconnected():
    raise Exception("无法连接到SAP HANA数据库")

cursor = conn.cursor()

# --------------------------
# 1. 获取需要导出的HANA表列表
# 替换为你的HANA模式名，比如'DEMO_SCHEMA'
hana_schema = '你的HANA模式名'
cursor.execute(f"""
    SELECT TABLE_NAME 
    FROM SYS.TABLES 
    WHERE SCHEMA_NAME = '{hana_schema}' 
      AND IS_USER_DEFINED = 'TRUE'  # 只导出用户自定义表，排除系统表
""")
tables_to_export = [row[0] for row in cursor.fetchall()]

# --------------------------
# 2. 遍历每个表完成导出
# BigQuery目标数据集（替换为你的项目和数据集）
bq_project = 'project1'
bq_dataset = 'test1'

for table_name in tables_to_export:
    print(f"开始处理表: {table_name}")
    
    try:
        # 读取HANA表数据
        cursor.execute(f"SELECT * FROM {hana_schema}.{table_name}")
        result = cursor.fetchall()
        
        # 获取表字段名（用于构建DataFrame列名）
        column_names = [desc[0] for desc in cursor.description]
        df = pd.DataFrame(result, columns=column_names)
        
        # 生成BigQuery Schema（自动推断数据类型）
        def get_bq_type(pd_dtype):
            if pd.api.types.is_integer_dtype(pd_dtype):
                return SqlTypeNames.INT64
            elif pd.api.types.is_float_dtype(pd_dtype):
                return SqlTypeNames.FLOAT64
            elif pd.api.types.is_datetime64_dtype(pd_dtype):
                return SqlTypeNames.DATETIME
            else:
                return SqlTypeNames.STRING
        
        schema = [
            SchemaField(col, get_bq_type(df[col].dtype)) 
            for col in df.columns
        ]
        
        # 定义BigQuery目标表ID
        bq_table_id = f"{bq_project}.{bq_dataset}.{table_name.lower()}"  # 转为小写，符合BigQuery命名习惯
        
        # 配置写入任务
        job_config = bigquery.LoadJobConfig(
            schema=schema,
            write_disposition="WRITE_TRUNCATE"  # 每次覆盖表数据，可改为WRITE_APPEND追加
        )
        
        # 写入BigQuery
        job = client.load_table_from_dataframe(df, bq_table_id, job_config=job_config)
        job.result()  # 等待任务完成
        
        # 验证写入结果
        table = client.get_table(bq_table_id)
        print(f"成功导出表 {table_name} 至 {bq_table_id}，共 {table.num_rows} 行数据")
        
    except Exception as e:
        print(f"处理表 {table_name} 失败: {str(e)}")
        continue

# 关闭数据库连接
conn.close()

注意事项 #

• HANA权限：确保DBADMIN用户有查询SYS.TABLES视图以及读取目标表数据的权限。
• 数据类型映射：上述代码的类型映射是基础版本，若有特殊数据类型（如HANA的DECIMAL、BOOLEAN），可扩展get_bq_type函数补充对应关系。
• 表名处理：BigQuery表名不支持大写和特殊字符，代码中转为小写，若有特殊需求可自定义命名转换规则。
• 错误处理：代码中加入了异常捕获，单个表处理失败不会中断整个批量任务。

内容的提问来源于stack exchange，提问作者Sandeep Mohanty