Section 1.1.1: Code Configuration for Job Bookmarks

To configure bookmarks at the code level, one must include “job.init” at the beginning, “job.commit()” at the end, and specify “transformation_ctx” for the various stages of the job, along with “jobBookmarkKeys” and “jobBookmarkKeysSortOrder” when invoking the source.

#!/usr/bin/env python

# -- coding: utf-8 --

# Importing libraries

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

# Params configuration

args = getResolvedOptions(sys.argv, ['JOB_NAME'])

sc = SparkContext()

glueContext = GlueContext(sc)

spark = glueContext.spark_session

job = Job(glueContext)

job.init(args['JOB_NAME'], args)

# Data source

datasource0 = glueContext.create_dynamic_frame

.from_catalog(database="database",

table_name="relatedqueries_csv",

transformation_ctx="datasource0",

additional_options={"jobBookmarkKeys":["col0"],

"jobBookmarkKeysSortOrder":"asc"})

# Data entries map

applymapping1 = ApplyMapping.apply(frame=datasource0,

mappings=[("col0", "long", "id", "long"),

("col1", "string", "number", "string"),

("col2", "string", "name", "string")

],

transformation_ctx="applymapping1")

# Data target

datasink2 = glueContext.write_dynamic_frame

.from_options(frame=applymapping1,

connection_type="s3",

connection_options={"path": "s3://input_path"},

format="json",

transformation_ctx="datasink2")

job.commit()

Using JDBC Connections

As previously mentioned, Glue jobs can source data from various repositories, with Amazon RDS and Amazon S3 being common choices. Bookmarks function effectively when ingesting data from files or objects stored in S3 buckets, regardless of whether records are new or have been updated since the last job execution. However, when jobs source data from RDS-hosted transactional databases, bookmarks may not capture records that have been modified post the last job run, even if they were initially captured.

Addressing this issue is straightforward, considering that AWS Glue operates on Spark. For instance, if a job reads from the “projects” table in an RDS database, which includes a timestamp field like “updated_at” that updates with each record modification, this field can serve as our incremental data load bookmark. We can create a Python function, like the one below, to check the last job run timestamp stored in a JSON file in the S3 bucket "my-ancillary-bucket" and formulate a SQL statement that filters records based on the bookmark.

#!/usr/bin/env python

# -- coding: utf-8 --

# Importing libraries

import boto3

import datetime

import json

import botocore

def checkingLastJobRunning(FileKey, jobBookmarkKeys):

'''

Function to check the last running date for the Glue job

Args:

• FileKey (string): Key for the object in S3 containing the last run timestamp
• jobBookmarkKeys (string): Name for the job bookmark key

Output:

• FilterStatement (string): Query statement

'''

# Checking current running timestamp

now_timestamp = datetime.datetime.utcnow()

.strftime("%Y-%m-%d %H:%M:%S.%f")

to_json = {

'previous_timestamp': now_timestamp

}

# Create a S3 client

s3 = boto3.resource('s3')

# Checking last running timestamp for the job

try:

content_object = s3.Object('my-ancillary-bucket', FileKey)

file_content = content_object.get()['Body'].read().decode('utf-8')

json_content = json.loads(file_content)

previous_timestamp = json_content['previous_timestamp']

symbol = '>'

except botocore.exceptions.ClientError:

previous_timestamp = now_timestamp

symbol = '<='

# Save new run timestamp

s3.Bucket('my-ancillary-bucket').put_object(Key=FileKey,

Body=(bytes(json.dumps(to_json).encode('utf-8'))))

# Filter statement

FilterStatement = f"WHERE {jobBookmarkKeys} ::timestamp {symbol} '{previous_timestamp}' :: timestamp"

return FilterStatement

Now, the above function can be utilized to formulate the SQL query that filters records from the "projects" table. By leveraging a JDBC connection and a Spark session (using a PostgreSQL driver in this example), a Spark DataFrame can be created containing entries made or altered since the last job execution.

#!/usr/bin/env python

# -- coding: utf-8 --

# Importing libraries

import sys

from pyspark.conf import SparkConf

from pyspark.context import SparkContext

from awsglue.utils import getResolvedOptions

from awsglue.context import GlueContext

from awsglue.job import Job

# Add additional spark configurations

conf = SparkConf() Any additional spark configuration you need --------------------------------------------------------

# Initial configuration

args = getResolvedOptions(sys.argv, ['JOB_NAME'])

glueContext = GlueContext(SparkContext.getOrCreate(conf=conf))

sparkSession = glueContext.spark_session

job = Job(glueContext)

job.init(args['JOB_NAME'], args)

# Checking filter statement

filter_statement = checkingLastJobRunning('my-previous-job-run.json',

'updated_at')

# RDS table to query

tablename = "projects"

# Data source

datasource0 = sparkSession.read.format("jdbc").

options(

url=f'jdbc:postgresql://{DB_HOST}:{DB_PORT}/{DB_NAME}',

dbtable=f"(SELECT * FROM {tablename} {filter_statement}) as subquery",

user=DB_USER,

password=DB_PASSWORD,

driver='org.postgresql.Driver').

load()