如何让AWS Glue爬虫/数据框正确识别日期字段并转换为RDS日期类型

1. 让AWS Glue爬虫自动识别日期字段 #

AWS Glue爬虫默认会将JSON中的日期字符串识别为字符串类型——因为JSON本身没有原生日期类型。不过可以通过以下方式让爬虫自动识别为日期类型：

• 使用自定义分类器：创建JSON分类器时，在配置中指定日期格式（如yyyy-MM-dd），爬虫会依据该规则将匹配的字符串字段识别为日期类型。
• 调整爬虫配置：在爬虫的“配置”选项中开启“推断日期/时间类型”功能（部分场景需结合自定义分类器生效）。
• 注意：你的示例JSON存在语法错误，键值对需用冒号而非等号，正确格式应为"date": "2024-07-09"，语法错误会直接影响爬虫的类型推断。

2. 在Spark DataFrame中将字符串日期转换为RDS兼容的日期类型 #

完全可以实现，具体步骤如下：

第一步：将字符串转换为Spark日期类型 #

使用Spark的to_date函数，将字符串字段转换为DateType，示例代码：

from pyspark.sql.functions import to_date, col

# 若日期为标准ISO格式（yyyy-MM-dd），可省略格式参数
data_frame = data_frame.withColumn("date", to_date(col("date"), "yyyy-MM-dd"))

第二步：写入RDS时自动映射为日期类型 #

当将DataFrame写入RDS（如MySQL、PostgreSQL）时，Spark会自动将DateType映射为对应数据库的日期类型（如MySQL的DATE、PostgreSQL的DATE）。只需确保目标表的对应字段为日期类型，并使用正确的JDBC驱动即可，示例写入代码：

# RDS连接配置
rds_url = "jdbc:mysql://your-rds-endpoint:3306/your-db"
rds_properties = {
    "user": "your-username",
    "password": "your-password",
    "driver": "com.mysql.cj.jdbc.Driver"
}

# 写入RDS表，mode可选append/overwrite等
data_frame.write.jdbc(
    url=rds_url,
    table="your-target-table",
    mode="append",
    properties=rds_properties
)

完整Glue ETL脚本示例 #

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 to_date, col

# 初始化上下文
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)

# 从Glue Catalog读取数据
data = glueContext.create_dynamic_frame.from_catalog(
    database="sample",
    table_name="table"
)

# 转换为Spark DataFrame
data_frame = data.toDF()

# 转换日期字段类型
data_frame = data_frame.withColumn("date", to_date(col("date")))

# 写入RDS
rds_url = "jdbc:mysql://your-rds-endpoint:3306/your-db"
rds_properties = {
    "user": "your-username",
    "password": "your-password",
    "driver": "com.mysql.cj.jdbc.Driver"
}
data_frame.write.jdbc(url=rds_url, table="target_table", mode="append", properties=rds_properties)

job.commit()

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