AWS Glue作业将S3数据迁移至Aurora时出现Data truncation报错

这个报错和异常JSON写入的问题，本质是AWS Glue读取CSV文件时触发了类型歧义识别：你表中的po_box_number列同时存在纯数字（比如100）和带字母的字符串（比如E101），Glue的DynamicFrame会默认将该列识别为choice混合类型，没有做强制类型转换的前提下，写入关系型数据库时会将整个混合类型结构体序列化成JSON字符串，自然超过了你最初设置的varchar(10)长度限制，触发数据截断报错。

1. 修正脚本逻辑错误 #

你当前脚本存在变量引用错误：ApplyMapping.apply的输入参数frame直接用了还未定义的applymapping1，需要先将数据源赋值给该变量。

2. 启用类型转换规则 #

你注释掉的ResolveChoice.apply方法就是Glue专门用来处理混合类型的API，取消注释即可强制将po_box_number列统一转换为字符串类型，避免结构体序列化。

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 datetime import datetime
from pyspark.sql.functions import lit
from awsglue.dynamicframe import DynamicFrame

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_catalog(database = "db1", table_name = "tb1", transformation_ctx = "datasource0")   

# 如果你需要用到AddProcessedTime逻辑可以取消下一行注释
# applymapping1 = Map.apply(frame = datasource0, f = AddProcessedTime)
# 没有额外Map处理的话直接用datasource0作为ApplyMapping的输入
applymapping1 = ApplyMapping.apply(frame = datasource0, mappings = [("col6", "string", "po_box_number", "string")], transformation_ctx = "applymapping1")

# 取消注释启用混合类型转字符串规则
applymapping1 = ResolveChoice.apply(applymapping1, specs = [("po_box_number", "cast:string")])

datasink5 = glueContext.write_dynamic_frame.from_catalog(frame = applymapping1, database = "db1", table_name = "tb2", transformation_ctx = "datasink5")
job.commit()

可选备选方案 #

如果上述方案还是存在类型转换问题，可以转为用Spark DataFrame强制指定字段类型：

# 插入在ResolveChoice之后，写入之前即可
df = applymapping1.toDF()
df = df.withColumn("po_box_number", df["po_box_number"].cast("string"))
applymapping1 = DynamicFrame.fromDF(df, glueContext, "fixed_frame")

内容的提问来源于stack exchange，提问作者parth222