Блог

Привет, я использую AWS Glue, чтобы попытаться загрузить данные из файла Json в S3 в Redshift. Я использую искатель Json с путем $ [ * ], и по какой-то причине одно из полей (оценка) входит в таблицу со структурой Json:

Есть какие-нибудь идеи о том, как сделать так, чтобы «оценка» просто отображала значение для самой оценки? Нужно ли мне настраивать скрипт PySpark для этой работы?

Пока что это сценарий:

 import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job
from pyspark.sql.functions import current_date 
from awsglue.dynamicframe import DynamicFrame

## @params: [TempDir, JOB_NAME]
args = getResolvedOptions(sys.argv, ['TempDir','JOB_NAME'])

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
## @type: DataSource
## @args: [database = "linkredshift", table_name = "uni3", transformation_ctx = "datasource0"]
## @return: datasource0
## @inputs: []
datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "linkredshift", table_name = "uni3", transformation_ctx = "datasource0")
df=datasource0.toDF().withColumn('data_date',current_date())  
datasource0 = DynamicFrame.fromDF(df, glueContext, "datasource0")

## @type: ApplyMapping
## @args: [mapping = [("first_name", "string", "first_name", "string"), ("last_name", "string", "last_name", "string"), ("subject", "string", "subject", "string"), ("grade", "string", "grade", "string")], transformation_ctx = "applymapping1"]
## @return: applymapping1
## @inputs: [frame = datasource0]
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("first_name", "string", "first_name", "string"), ("last_name", "string", "last_name", "string"), ("subject", "string", "subject", "string"), ("grade", "string", "grade", "string"), ("data_date", "date", "data_date", "date")], transformation_ctx = "applymapping1")
## @type: SelectFields
## @args: [paths = ["subject", "grade", "last_name", "first_name"], transformation_ctx = "selectfields2"]
## @return: selectfields2
## @inputs: [frame = applymapping1]
selectfields2 = SelectFields.apply(frame = applymapping1, paths = ["subject", "grade", "last_name", "first_name", "data_date"], transformation_ctx = "selectfields2")
## @type: ResolveChoice
## @args: [choice = "MATCH_CATALOG", database = "linkredshift", table_name = "dev_public_students", transformation_ctx = "resolvechoice3"]
## @return: resolvechoice3
## @inputs: [frame = selectfields2]
resolvechoice3 = ResolveChoice.apply(frame = selectfields2, choice = "MATCH_CATALOG", database = "linkredshift", table_name = "dev_public_students", transformation_ctx = "resolvechoice3")
## @type: ResolveChoice
## @args: [choice = "make_cols", transformation_ctx = "resolvechoice4"]
## @return: resolvechoice4
## @inputs: [frame = resolvechoice3]
resolvechoice4 = ResolveChoice.apply(frame = resolvechoice3, choice = "make_cols", transformation_ctx = "resolvechoice4")
## @type: DataSink
## @args: [database = "linkredshift", table_name = "dev_public_students", redshift_tmp_dir = TempDir, transformation_ctx = "datasink5"]
## @return: datasink5
## @inputs: [frame = resolvechoice4]
datasink5 = glueContext.write_dynamic_frame.from_catalog(frame = resolvechoice4, database = "linkredshift", table_name = "dev_public_students", redshift_tmp_dir = args["TempDir"], transformation_ctx = "datasink5")
job.commit()
 

Поскольку поле grade может быть строкой или значением int, выбор решения с приведением к нему необходимо сделать одним или другим. Попробуйте изменить строку applymapping1 на эти две строки:

 datasource0= datasource0.resolveChoice(specs=[("grade",'cast:int')])
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("first_name", "string", "first_name", "string"), ("last_name", "string", "last_name", "string"), ("subject", "string", "subject", "string"), ("grade", "int", "grade", "int"), ("data_date", "date", "data_date", "date")], transformation_ctx = "applymapping1")
 

Обратите внимание, что это изменяет ваше поле оценки на тип int (как это то, что я выбираю в выборе решения, чтобы использовать значения в int). Не стесняйтесь преобразовывать поле оценка в строку вместо этого позже, но int кажется лучшим выбором.

Смотрите здесь для справки: https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-python-samples-medicaid.html