如何在 AWS Glue 中从 CSV 创建结构化 JSON

Question

我能够以预期的格式创建结构化 json，但额外的斜杠出现在 JSON 记录中，并且 json 记录显示为字符串对象。

请详细说明解决方案或让我知道缺少什么或是否存在任何其他方法来实现预期结果。

我目前的结果：

{
   "awsservices":[
      "{"\key":\"string_value"\, \"key":\numeric_value, "\key":\"amazon\web/services"}",
      "{"\key":\"string_value"\, \"key":\numeric_value, "\key":\"amazon\web/services"}",
      "{"\key":\"string_value"\, \"key":\numeric_value, "\key":\"amazon\web/services"}",
      "{"\key":\"string_value"\, \"key":\numeric_value, "\key":\"amazon\web/services"}"
   ]
} 

预期结果：

{
   "awsservices":[
      {"key":"string_value", "key":numeric_value, "key":"amazon web services"},
      {"key":"string_value", "key":numeric_value, "key":"amazon web services"},
      {"key":"string_value", "key":numeric_value, "key":"amazon web services"},
      {"key":"string_value", "key":numeric_value, "key":"amazon\web/services"}
   ]
} 

我的代码：

SourceDataDYF = glueContext.create_dynamic_frame.from_options(
   format_options = {"quoteChar": '"', "escaper":"", "withHeader":True, "separator":"|", "inferSchema":"false"},
   connection_type = "s3",
   format = "csv",
   connection_options = {"paths": "s3:bucket_name/csv_file_path/"], "recurse":True},
   transformation_ctx = "SourceDataDYF"
)

StageDataDF = SourceDataDYF.toDF()

print("*******************************: WRITE JSON :*******************************")

PreStageDataDF1 = StageDataDF.select(to_json(struct(*StageDataDF.columns)).alias("json")) \
   .groupBy(spark_partition_id()) \
   .agg(collect_list("json").alias("awsservices")) \
   .select(col("awsservices").cast("string")).coalesce(1)

targetDataDYF = DynamicFrame.fromDF(PreStageDataDF1,glueContext,"PreStageDataDF1")
targetDataJSON = glueContext.write_dynamic_frame.from_options(
   frame = targetDataDYF,
   connection_type = "s3",
   connection_options = {"path": "s3://result_bucket_name/folder_path/", "partitionKeys": []},
   format = "json",
   transformation_ctx = "targetDataJSON"
)

Answer 1

我使用以下代码创建了结构化或预期格式的 JSON：

StageDataDF = SourceDataDYF.toDF()
StageDataDF.schema
StageDataDF.show(5)

print("**************:Converting Spark DF to pandas DF:******************")
StageDataDF1 = StageDataDF.toPandas()
print(StageDataDF1)

print("#############: Converting into Python Dictionary:#############")
StageDataDF2 = StageDataDF1.to_dict(orient = 'records')
print(StageDataDF2)

print("#############: Converting Dictionary to 2D list:#############")
StageDataDF3 =[[ rec for rec in StageDataDF2 ]]
StageDataDF4 = {"awsservices":StageDataDF3}
print(StageDataDF4)

print("#############: Converting Python Dictionary to Pandas DataFrame:#############")
StageDataDF5 = pd.DataFrame(data = StageDataDF4)
print(type(StageDataDF5))

print("#############: Converting Pandas DataFrame to Spark DF again:#############")
StageDataDF6 = spark.createDataFrame(StageDataDF5)
print(type(StageDataDF6))
StageDataDF6.show()

print("*******************************: Traget Dynamic Data Frame :*******************************")
targetDataDYF = DynamicFrame.fromDF(StageDataDF6, glueContext, "StageDataDF6").coalesce(1)
print(type(targetDataDYF))
targetDataDYF.show()

targetDataJSON = glueContext.write_dynamic_frame.from_options(frame = targetDataDYF, connection_type = "s3", connection_options = {"path": "s3://bucket/folder/", "partitionKeys": []},format = "json",transformation_ctx = "targetDataJSON")