使用 AWS Glue 或 PySpark 过滤 DynamicFrame

我的 AWS Glue 数据目录中有一个名为“mytable”的表。该表位于本地 Oracle 数据库连接“mydb”中。

我想将生成的 DynamicFrame 过滤为仅 X_DATETIME_INSERT 列（时间戳）大于特定时间（在本例中为“2018-05-07 04:00:00”）的行。之后，我尝试对行进行计数以确保计数较低（该表大约有 40,000 行，但只有几行应该满足过滤条件）。

这是我当前的代码：

import boto3
from datetime import datetime
import logging
import os
import pg8000
import pytz
import sys
from awsglue.context import GlueContext
from awsglue.job import Job
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from base64 import b64decode
from pyspark.context import SparkContext
from pyspark.sql.functions import lit
## @params: [TempDir, JOB_NAME]
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)

datasource0 = glueContext.create_dynamic_frame.from_catalog(database = "mydb", table_name = "mytable", transformation_ctx = "datasource0")

# Try Glue native filtering    
filtered_df = Filter.apply(frame = datasource0, f = lambda x: x["X_DATETIME_INSERT"] > '2018-05-07 04:00:00')
filtered_df.count()

此代码运行 20 分钟并超时。我尝试过其他变体：

df = datasource0.toDF()
df.where(df.X_DATETIME_INSERT > '2018-05-07 04:00:00').collect()

And

df.filter(df["X_DATETIME_INSERT"].gt(lit("'2018-05-07 04:00:00'")))

哪些失败了。我究竟做错了什么？我在 Python 方面经验丰富，但对 Glue 和 PySpark 很陌生。

AWS Glue 将整个数据集从 JDBC 源加载到临时 s3 文件夹中，然后应用过滤。如果您的数据位于 s3 而不是 Oracle 中，并按某些键分区（即/年/月/日），那么您可以使用下推谓词特征 https://aws.amazon.com/blogs/big-data/work-with-partitioned-data-in-aws-glue/加载数据子集：

val partitionPredicate = s"to_date(concat(year, '-', month, '-', day)) BETWEEN '${fromDate}' AND '${toDate}'"

val df = glueContext.getCatalogSource(
   database = "githubarchive_month",
   tableName = "data",
   pushDownPredicate = partitionPredicate).getDynamicFrame()

不幸的是，这还不适用于 JDBC 数据源。