使用AWS Glue/PySpark过滤DynamicFrame超时及失败问题求助

首先，我帮你梳理下之前几种写法的问题所在，再给出针对性的解决方案：

为什么你的代码会失败/超时？ #

1. Glue原生Filter.apply的问题：
   你用Python lambda逐行过滤的方式，是在数据全量拉取到Glue后，在Driver端逐行处理，效率极低。而且直接把Timestamp类型字段和字符串比较，即便Spark会做隐式转换，这种方式也完全没利用数据库的过滤能力，导致4万行数据全部传输到Glue后再过滤，自然容易超时。

2. Spark DataFrame的collect()问题：
   collect()会把过滤后的所有结果拉到Driver内存中，哪怕符合条件的是少数行，这种操作也没必要；而且如果隐式类型转换出问题，过滤逻辑也会直接失效。

3. 第三种写法的语法错误：
   lit("'2018-05-07 04:00:00'")里多了一层单引号，相当于拿带引号的字符串和Timestamp类型字段比较，类型完全不匹配，肯定过滤不出任何结果。

最优解决方案：让过滤条件下推到Oracle执行 #

这是解决超时最关键的一步——不要让Glue拉全表数据，而是直接让Oracle数据库执行过滤，只返回符合条件的行。Glue支持通过push_down_predicate参数实现这一点：

import sys
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext

## 初始化Glue上下文和Job
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)

# 重点：编写Oracle原生的过滤条件，用TO_TIMESTAMP处理时间字符串
predicate = "X_DATETIME_INSERT > TO_TIMESTAMP('2018-05-07 04:00:00', 'YYYY-MM-DD HH24:MI:SS')"

# 读取数据时直接传入push_down_predicate，让Oracle先过滤
datasource0 = glueContext.create_dynamic_frame.from_catalog(
    database = "mydb", 
    table_name = "mytable", 
    transformation_ctx = "datasource0",
    push_down_predicate = predicate
)

# 直接统计行数，此时数据量已经很小了
print(f"符合条件的行数：{datasource0.count()}")

job.commit()

备选方案：转Spark DataFrame后正确过滤 #

如果你一定要转成DataFrame处理，要注意显式类型转换和避免用collect()：

from pyspark.sql.functions import to_timestamp, lit

# 先转成DataFrame
df = datasource0.toDF()

# 显式将字符串转为Timestamp类型，确保类型匹配
filter_time = to_timestamp(lit("2018-05-07 04:00:00"), "yyyy-MM-dd HH:mm:ss")

# 执行过滤，用count()而不是collect()统计结果
filtered_df = df.filter(df.X_DATETIME_INSERT > filter_time)
print(f"符合条件的行数：{filtered_df.count()}")

额外提醒 #

• 永远优先用push_down_predicate下推过滤条件到源数据库，这能大幅减少数据传输量，提升性能。
• 处理时间类型时，尽量显式转换，避免依赖Spark的隐式转换，减少类型不匹配的问题。
• 统计行数用count()即可，不要用collect()，除非你真的需要把所有结果拉到本地处理。

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