У меня есть столбец, который больше, чем тип данных varchar (max), который, насколько я понимаю, является самым большим типом данных, который использует AWS Glue, и получаю сообщение об ошибке «Длина строки превышает длину DDL», когда я пытаюсь загрузить свои таблицы из-за Это. Я не пытаюсь обрезать этот столбец, потому что он не так уж важен и не могу понять, как это сделать в Glue. Я знаю, что могу использовать TRUNCATECOLUMNS в качестве тега в команде копирования, если я подключаюсь к своей базе данных с помощью psql в экземпляре EC2 и действительно могу успешно загрузить свои таблицы таким образом. Однако мой босс настаивает, чтобы я использовал Glue для этой работы, поэтому я ищу способ обрезать столбцы с помощью скрипта Glue. Я просмотрел много документации, но не нашел ничего похожего. Спасибо.
Вот рабочий код для всех, кто может иметь эту проблему и нуждается в полной справке. Обратите внимание, что varchar(65535) - это максимальное количество символов, которое столбец может быть в Redshift:
val truncColUdf = udf((str: String) => if (str.length > 29999) str.substring(0, 29999) else str)
val datasource30 = glueContext.getCatalogSource(database = "database", tableName = "entry", redshiftTmpDir = "", transformationContext = "datasource30").getDynamicFrame()
val revDF30 = datasource30.toDF()
.withColumn("message", truncColUdf(col("message")))
val truncDynamicFrame30 = DynamicFrame(revDF30, glueContext)
val applymapping30 = truncDynamicFrame30.applyMapping(mappings = Seq(("id", "bigint", "id", "bigint"), ("message", "string", "message", "varchar(65535)"), ("state", "string", "state", "varchar(256)"), ("created_at", "timestamp", "created_at", "timestamp"), ("depth", "int", "depth", "int")), caseSensitive = false, transformationContext = "applymapping30")
val resolvechoice30 = applymapping30.resolveChoice(choiceOption = Some(ChoiceOption("make_cols")), transformationContext = "resolvechoice30")
val dropnullfields30 = resolvechoice30.dropNulls(transformationContext = "dropnullfields30")
val datasink30 = glueContext.getJDBCSink(catalogConnection = "databaseConnection", options = JsonOptions("""{"dbtable": "entry", "database": "database"}"""), redshiftTmpDir = args("TempDir"), transformationContext = "datasink30").writeDynamicFrame(dropnullfields30)
Вот пример строки считываемых данных:
01,"<p>Here is the message where the quotations are in case of commas within the message, like so.</p>",active,2017-08-27 23:38:40,1
