AWS Glue Job — загрузить файл паркета из S3 в столбец RDS jsonb

У меня есть файл паркета в S3, в котором есть несколько столбцов, и один из них - json. У меня такой же формат в базе данных RDS с одним столбцом, как jsonb.

Я хотел бы скопировать файл паркета в RDS, но как преобразовать файл в тип данных jsonb, поскольку Glue не поддерживает тип столбца json. Когда я пытаюсь вставить столбец как строку, я получаю сообщение об ошибке. Любые идеи о том, как я могу ввести столбец json в столбец RDS jsonb?

 An error occurred while calling o145.pyWriteDynamicFrame. ERROR: column "json_column" is of type jsonb but expression is of type character varyin
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

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

sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)

DataSource0 = glueContext.create_dynamic_frame.from_options(connection_type = "s3", format = "parquet", connection_options = {"paths": ["s3://folder"], "recurse":True}, transformation_ctx = "DataSource0")
Transform0 = ApplyMapping.apply(frame = DataSource0, mappings = [("id", "long", "id", "long"), ("name", "string", "name", "string"), ("json_column", "string", "json_column", "string")], transformation_ctx = "Transform0")

DataSink0 = glueContext.write_dynamic_frame.from_catalog(frame = Transform0, database = "postgres", table_name = "table", transformation_ctx = "DataSink0")
job.commit()

person needtocode    schedule 12.04.2021    source источник


Ответы (1)


Один из способов — подключиться к вашему RDS с помощьюPsychopg2, перебрать ваш набор данных и загрузить его напрямую.

Как вставить JSONB в Postgresql с помощью Python?

person Coin Graham    schedule 12.04.2021