this are 5 different code snippets that i tried for performance comparison, only 2 actually filtered data on the server level when using profiler, it seems at the moment without creating a custom connector or buying from marketplace the only way to get this to work is using glueContext.read
You can convert DynamicFrames to and from DataFrames (See example)
rds_datasink_temp = DynamicFrame.fromDF(rds_dataframe, glueContext, "nested")
you should also check this while running Sql Server Profiler with all the events from: OLEDB, Stored Procedure, TSQL and Transactions
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 awsglue.context import DynamicFrame
# list parameter with 2 leading hyphens --param_server_url
args = getResolvedOptions(sys.argv,['JOB_NAME'])
print("JOB_NAME: ", args['JOB_NAME'])
job_server_url="SERVER URL"
job_db_name="DB NAME"
job_db_user="DB USER"
job_db_password="DB PASSWORD"
job_table_name="TABLE NAME"
job_glue_db_name="GLUE DATA CATALOG DATABASE NAME"
job_glue_conn_name="GLUE DATA CATALOG CONNECTION NAME"
job_glue_table_name="GLUE DATA CATALOG TABLE NAME"
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)
region = "us-east-1"
#### aws glue data catalog table info (from ) ####
# Name job_glue_table_name
# Database job_glue_db_name
# Classification sqlserver
# Location job_db_name.dbo.job_table_name
# Connection job_glue_conn_name
#### GlueContext Class ####
# https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-crawler-pyspark-extensions-glue-context.html
#### DynamicFrame Class ####
# https://docs.aws.amazon.com/glue/latest/dg/aws-glue-api-crawler-pyspark-extensions-dynamic-frame.html
#### Connection Api ####
# https://docs.aws.amazon.com/glue/latest/webapi/API_Connection.html
#### Using connectors and connections with AWS Glue Studio ####
# Link : https://docs.aws.amazon.com/glue/latest/ug/connectors-chapter.html
# Use AWS Secrets Manager for storing credentials
# Filtering the source data with row predicates and column projections
#### Connection options for type custom.jdbc or marketplace.jdbc ####
# Link : https://docs.aws.amazon.com/glue/latest/dg/aws-glue-programming-etl-connect.html#aws-glue-programming-etl-connect-jdbc
# className – String, required, driver class name.
# connectionName – String, required, name of the connection that is associated with the connector.
# secretId or user/password – String, required, used to retrieve credentials for the URL.
# dbTable or query – String, required, the table or SQL query to get the data from. You can specify either dbTable or query, but not both.
# filterPredicate – String, optional, extra condition clause to filter data from source. For example:
# using \ for new line with more commands
# query="recordid<=5", -- filtering !
print("0001 - df_read_query")
df_read_query = glueContext.read \
.format("jdbc") \
.option("url","jdbc:sqlserver://"+job_server_url+":1433;databaseName="+job_db_name+";") \
.option("query","select recordid from "+job_table_name+" where recordid <= 5") \
.option("user",job_db_user) \
.option("password",job_db_password) \
.load()
print("df_read_query count: ", df_read_query.count())
df_read_query.show(10)
df_read_query.printSchema()
# query="recordid<=5", -- not filtering
print("0002 - df_from_catalog_query")
df_from_catalog_query = glueContext.create_dynamic_frame.from_catalog(
database = job_glue_db_name,
table_name = job_glue_table_name,
additional_options={
"query":"select recordid from "+job_table_name+" where recordid <= 5;",
},
transformation_ctx = "df_from_catalog_query",
)
print("df_from_catalog_query count: ", df_from_catalog_query.count())
df_from_catalog_query.show(10)
# push_down_predicate="recordid<=5", -- not filtering
print("0003 - df_from_catalog_push_down_predicate")
df_from_catalog_push_down_predicate = glueContext.create_dynamic_frame.from_catalog(
database = job_glue_db_name,
table_name = job_db_name+'_dbo_'+job_table_name,
push_down_predicate = "recordid<=5",
transformation_ctx = "df_from_catalog_push_down_predicate",
)
print("df_from_catalog_push_down_predicate count: ", df_from_catalog_push_down_predicate.count())
df_from_catalog_push_down_predicate.show(10)
# filterPredicate="recordid<=5", -- not filtering
print("0004 - df_from_options_sqlserver")
df_from_options_sqlserver = glueContext.create_dynamic_frame.from_options(
connection_type = "sqlserver",
connection_options = {
"url":"jdbc:sqlserver://"+job_server_url+":1433;databaseName="+job_db_name+";",
"username":job_db_user,
"password":job_db_password,
"location":job_db_name+".dbo."+job_table_name,
"filterPredicate":"recordid<=5",
},
transformation_ctx = "df_from_options_sqlserver",
)
print("df_from_options_sqlserver count: ", df_from_options_sqlserver.count())
df_from_options_sqlserver.show(10)
# dbtable="recordid<=5", -- filtering !
print("0005 - df_read_dbtable")
df_read_dbtable = glueContext.read \
.format("jdbc") \
.option("url","jdbc:sqlserver://"+job_server_url+":1433;databaseName="+job_db_name+";") \
.option("user",job_db_user) \
.option("password",job_db_password) \
.option("dbtable","(select recordid from "+job_table_name+" where recordid<=5) as t1") \
.option("driver","com.microsoft.sqlserver.jdbc.SQLServerDriver") \
.load()
print("df_read_dbtable count: ", df_read_dbtable.count())
df_read_dbtable.show(10)
job.commit()