recommend-emr-dataprocess/ad/hive_test.py

from odps import ODPS
import pandas as pd


def get_odps_instance(project):
    odps = ODPS(
        access_id='LTAIWYUujJAm7CbH',
        secret_access_key='RfSjdiWwED1sGFlsjXv0DlfTnZTG1P',
        project=project,
        endpoint='http://service.cn.maxcompute.aliyun.com/api',
    )
    return odps

if __name__ == '__main__':
    project = 'pai_algo'
    odps = get_odps_instance(project)
    table = 'pai_temp_flow_1p2ujuopb682ktq1h6_node_uxam9b0n7k2v11i9rz_outputTable'
    num = 10
    try:
        # 要查询的 SQL 语句
        sql = f'select * from {table} limit {num}'
        # 执行 SQL 查询
        with odps.execute_sql(sql).open_reader() as reader:
            # 查询数量小于目标数量时 返回空
            if reader.count < num:
                print("error")
            # 获取字段名称
            column_names = [col.name for col in reader._schema.columns]
            # 获取查询结果数据
            data = []
            for record in reader:
                record_list = list(record)
                numbers = []
                for item in record_list:
                    numbers.append(item[1])
                data.append(numbers)
            # 将数据和字段名称组合成 DataFrame
            df = pd.DataFrame(data, columns=column_names)
            print(df)
    except Exception as e:
        print(f"发生错误: {e}")