AWS Glue - 将来自 GET(REST API) 请求的 Json 响应转换为 DataFrame/DyanamicFramce 并将其存储在 s3 存储桶中

如何解决AWS Glue - 将来自 GET(REST API) 请求的 Json 响应转换为 DataFrame/DyanamicFramce 并将其存储在 s3 存储桶中

headersAPI = {
     'Content-Type': 'application/json','accept': 'application/json','Authorization': 'Bearer XXXXXXXXXXXXXXXXXXXXXXXXXX',}
skill_response=requests.get("XXXXXX",headers=headersAPI),headers=headersAPI)

log.info(skill_response.text)
skill_json=skill_response.json()
print(skill_json)  ##print the json data and verified
    
log.info('skills data')
log.info(skill_json["status"]) 
        
DataSink0 = glueContext.write_dynamic_frame.from_options(frame =
   skill_json,connection_type = "s3",format = "csv",connection_options=
   {"path": "s3://xxxxx/","partitionKeys": []},transformation_ctx= "DataSink0")

job.commit()

类型错误：frame_or_dfc 必须是 DynamicFrame 或DynamicFrameCollection。得到

在写入 S3 时出现此错误：'dict' object has no attribute '_jdf'

解决方法

通过首先从响应字符串创建一个 DataFrame（讨论 here），然后将此 DataFrame 转换为 DynamicFrame，可以将 JSON 响应转换为 DynamicFrame。

这个例子应该可以工作：

import requests
from awsglue.job import Job
from pyspark.context import SparkContext

from awsglue import DynamicFrame
from awsglue.context import GlueContext

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)

r = requests.get(url='https://api.github.com/users?since=100')

df = spark.read.json(sc.parallelize([r.text]))

dynamic_frame = DynamicFrame.fromDF(
    df,glue_ctx=glueContext,name="df"
)

#dynamic_frame.show()

DataSink0 = glueContext.write_dynamic_frame.from_options(
    frame=dynamic_frame,connection_type="s3",format="csv",connection_options={"path": "s3://xxxxx/","partitionKeys": []},transformation_ctx="DataSink0")
job.commit()