AWS Glue script showing how to avoid duplicates during a job execution.

aws_glue_avoiding_duplicates.py

import sys
import pydevd
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.dynamicframe import DynamicFrame
from pyspark.sql.functions import col
from awsglue.job import Job


def main():

    # Invoke pydevd to remote debug
    pydevd.settrace('169.254.76.0', port=9001, stdoutToServer=True, stderrToServer=True)

    args = getResolvedOptions(sys.argv, ['JOB_NAME'])
    sc = SparkContext()
    gc = GlueContext(sc)
    job = Job(gc)
    job.init("MyJob", args)

    stagingdatasource = gc.create_dynamic_frame.from_catalog(
        database="stagingdatabase",
        table_name="staging_source_table",
        transformation_ctx="stagingdatasource")

    targetdatasource = gc.create_dynamic_frame.from_catalog(
        database="targetdatabase",
        redshift_tmp_dir=args["TempDir"],
        table_name="target_table",
        transformation_ctx="targetdatasource")

    columnmapping = ApplyMapping.apply(
        frame=stagingdatasource,
        mappings=[("description", "string", "description", "string"), ("id", "int", "id", "int")],
        transformation_ctx="columnmapping")

    ta = columnmapping.toDF().alias('ta')
    tb = targetdatasource.toDF().alias('tb')

    left_join = ta\
        .join(tb, ta.value == tb.value, how='left')\
        .filter(col('tb.value').isNull())\
        .select('ta.*')

    # Inspect left join
    # left_join.show()

    finaldf = DynamicFrame.fromDF(left_join, gc, "nested")

    gc.write_dynamic_frame.from_catalog(
        frame=finaldf,
        database="targetdatabase",
        redshift_tmp_dir=args["TempDir"],
        table_name="target_table")

    job.commit()


if __name__ == "__main__":
    main()

Hey @michelmilezzi, what does ta.value represent? When I tried to implement this I get an error that the dataframe has no attribute value