AWS Glue用Python Shell从Workday系统将人力资源原始数据以Parquet格式存入S3

使用AWS Glue Python Shell从Workday获取人力资源数据并存储到S3的详细方案：

通过以下步骤，即可实现从Workday到S3的自动化数据管道。实际部署时需根据Workday API的具体响应结构调整数据处理逻辑。

架构流程

Workday API -> AWS Glue Python Shell -> (Parquet) -> Amazon S3

---

详细步骤

一、前期准备

1. 获取Workday API访问权限

   • 确认Workday REST/SOAP API端点
   • 获取API认证凭证（OAuth2令牌/Basic Auth/证书）
   • 确认所需人力资源数据的具体API路径（如/human_resources/employees）

2. 创建S3存储桶

   • 创建目标桶（如s3://my-hr-data-bucket/raw/）

3. 创建IAM角色

   • 创建Glue服务角色（如GlueHRDataRole）
   • 附加策略：

     {"Version": "2012-10-17","Statement": [{"Effect": "Allow","Action": ["s3:PutObject","s3:GetObject"],"Resource": "arn:aws:s3:::my-hr-data-bucket/*"},{"Effect": "Allow","Action": ["logs:CreateLogGroup","logs:CreateLogStream","logs:PutLogEvents"],"Resource": "*"}]
     }
     

---

二、Python脚本开发

# glue_workday_to_s3.py
import requests
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import os
from datetime import datetime
import logging# 初始化日志
logger = logging.getLogger()
logger.setLevel(logging.INFO)# Workday配置
WORKDAY_API_URL = "https://api.workday.com/v1/human_resources"
API_KEY = "your_workday_api_key"
HEADERS = {"Authorization": f"Bearer {API_KEY}","Accept": "application/json"
}# S3配置
S3_BUCKET = "my-hr-data-bucket"
S3_PREFIX = "raw/hr_data/"
FILE_NAME = f"hr_data_{datetime.today().strftime('%Y%m%d')}.parquet"def fetch_paginated_data(url):"""处理API分页"""all_data = []page = 1while True:response = requests.get(f"{url}?page={page}",headers=HEADERS)if response.status_code != 200:logger.error(f"API请求失败: {response.text}")breakdata = response.json()all_data.extend(data['items'])if data['has_more']:page +=1else:breakreturn all_datadef main():try:# 获取数据hr_data = fetch_paginated_data(f"{WORKDAY_API_URL}/employees")logger.info(f"获取到 {len(hr_data)} 条记录")# 转换为DataFramedf = pd.DataFrame(hr_data)# 转换为Parquettable = pa.Table.from_pandas(df)output_path = f"/tmp/{FILE_NAME}"pq.write_table(table, output_path)# 上传到S3s3_client = boto3.client('s3')s3_client.upload_file(output_path,S3_BUCKET,f"{S3_PREFIX}{FILE_NAME}")logger.info("数据成功写入S3")except Exception as e:logger.error(f"处理失败: {str(e)}")raiseif __name__ == "__main__":main()

---

三、Glue作业配置

1. 创建Python Shell作业

   • 作业名称：workday-hr-data-ingestion
   • IAM角色：选择GlueHRDataRole
   • 类型：Python Shell
   • Python版本：Python 3.9
   • 数据处理单元：1/16 DPU

2. 作业参数

   • 脚本路径：上传glue_workday_to_s3.py到S3并指定路径
   • 添加Python库依赖：

     --additional-python-modules pandas==1.5.3,pyarrow==12.0.1,requests==2.28.2
     

   • 环境变量（可选）：

     --WORKDAY_API_KEY=your_actual_key
     

3. 设置触发器

   • 按需运行 或 使用EventBridge定时触发（如每天凌晨1点）

---

四、验证数据

1. 在S3目标路径检查Parquet文件：

   aws s3 ls s3://my-hr-data-bucket/raw/hr_data/
   

2. 使用Athena验证数据：

   CREATE EXTERNAL TABLE hr_data (employee_id string,name string,department string,...
   )
   STORED AS PARQUET
   LOCATION 's3://my-hr-data-bucket/raw/hr_data/';
   

---

注意事项

1. 敏感信息管理
   • 建议将API密钥存储在AWS Secrets Manager，通过boto3动态获取
2. 错误处理
   • 添加重试逻辑应对API限流
   • 使用Glue Job Bookmarks跟踪处理进度（可选）
3. 性能优化
   • 若数据量极大（>1GB），建议改用Glue Spark作业
4. 成本控制
   • Python Shell作业成本低于Spark作业，适合中等数据量