目录

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_listprint(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