【Python】 -- python3 读取 aws athena 表数据

目录

1、环境准备

2、安装环境

3、举例查询某张表数据和执行 add partition 操作

3.1、编辑文件 athena_jdbc.py

3.2、查找 JVM 的动态链接库路径

3.3、保存文件，执行以下命令

1、环境准备

• oracle jdk 11
• centos 8
• 依赖：pandas、pyathenajdbc 和 sqlalchemy

2、安装环境

• $ sudo pip install pandas -y
• $ sudo pip install pyathenajdbc -y
• $ sudo pip install sqlalchemy -y

3、举例查询某张表数据和执行 add partition 操作

3.1、编辑文件 athena_jdbc.py

import time
from datetime import datetime
import pandas as pd
from pyathenajdbc import connect
from sqlalchemy import create_engine, text


# 配置 AWS Athena 的连接参数
aws_access_key_id = "your_aws_access_key_id"
aws_secret_access_key = "your_aws_secret_access_key"
aws_region = "your_region"
s3_output_location = "s3://your-test-bucket/athena_result/"
work_group = "primary"


"""
获取今天的日期并格式化为 YYYYMMDD 的字符串格式。

Returns:
    str: 格式化后的日期字符串。
"""
def get_formatted_date():
    today = datetime.today()
    formatted_date = today.strftime("%Y%m%d")
    return formatted_date



# 获取读取表数据的链接
conn = connect(User='aws_access_key_id',
               Password='aws_secret_access_key',
               S3OutputLocation='s3://your-test-bucket/athena_result',
               AwsRegion='your_region',
               jvm_path='/usr/lib/jvm/java-11/lib/server/libjvm.so')


# 获取执行 ddl 的链接
engine = create_engine(
    f"awsathena+rest://{aws_access_key_id}:{aws_secret_access_key}@athena.{aws_region}.amazonaws.com:443/"
    f"default?s3_staging_dir={s3_output_location}&work_group={work_group}"
)


"""
生成多张表的 athena 添加分区 SQL。

Args:
    today_time (str): 当前日期，格式为 'yyyyMMdd'。
    table_names (list): 表名列表。

Returns:
    list: 包含所有表的 athena SQL 语句的列表。
"""
def generate_athena_partition_sqls(data_date, table_names):
    sql_list = []
    for table_name in table_names:
        sql = (
            f"ALTER TABLE tg_bigdata.{table_name} "
            f"ADD IF NOT EXISTS PARTITION (data_date=\'{data_date}\') "
            f"LOCATION \'s3://your-test-bucket//ods/{table_name}/data_date={data_date}/\';"
        )
        sql_list.append(sql)
    
    return sql_list

print(f"------> Creating SQL...")
data_date = get_formatted_date()
table_names = ["ods_user_all", "ods_music_all"]
exec_sqls = generate_athena_partition_sqls(data_date, table_names)

print(f"------> Created SQL...")
print(f"------> Starting executing SQL...")
with engine.connect() as conn:
    for sql in exec_sqls:
        conn.execute(text(sql))
        print(f"Executed SQL: {sql}...")

print(f"------> Ended executing SQL...")

3.2、查找 JVM 的动态链接库路径

$ sudo find /usr/lib/jvm -name "libjvm.so"
修改 jvm_path 为查到到的地址。

3.3、保存文件，执行以下命令

注意：执行前，需要添加环境变量！

export AWS_ACCESS_KEY_ID=your_aws_access_key_id
export AWS_SECRET_ACCESS_KEY=aws_secret_access_key
export AWS_DEFAULT_REGION=your_region
export AWS_ATHENA_S3_STAGING_DIR=s3://your-test-bucket/athena_result/jdbc

$ python3 athena_jdbc.py