Peoplesoft to Salesforce: Education History Sync

#   _____           _              _____        __      
#  |  __ \         (_)            |_   _|      / _|     
#  | |__) |___  ___ _ _ __   ___    | |  _ __ | |_ ___  
#  |  _  // _ \/ __| | '_ \ / _ \   | | | '_ \|  _/ _ \ 
#  | | \ \  __/ (__| | |_) |  __/  _| |_| | | | || (_) |
#  |_|  \_\___|\___|_| .__/ \___| |_____|_| |_|_| \___/ 
#                    | |                                
#                    |_|

# Project Name - Peoplesoft to Salesforce: Education History Sync

# Recipe URL - https://app.lingk.io/a/10932/tf/19103

# Description - 10 - Peoplesoft to Salesforce: Education History Sync
                # Select all education history data from PeopleSoft id student exist in PS_SALFM_ADM_STDT and PS_EXT_DEGREE PS tables.
                # Collect info from all tables - EXT_ACAD_DATA, EXT_ACAD_SUM, EXT_DEGREE and transform fields.
                # Select Account and Contact from salesforce using EXT_ORG_ID and EMPLID.
                # If match found in SF PeopleSoft_ACAD_HISTORY_EXT_ORG_ID__c then update else insert record.
                # Insert education history requested fields with their SF id in Lingk environment file.

# Systems - Peoplesoft, Salesforce
# Connectors - HttpV2, Salesforce
# Data Flows - Single Direction
# Connection Type - API to API

# Data Models
## Peoplesoft Models Used:
#1. Person
#2. EXT_ACAD_DATA

## Salesforce Objects Used:
#1. Accounts

# Add Recipe notes / Change log information here!

#    _____                            _                 
#   / ____|                          | |                
#  | |     ___  _ __  _ __   ___  ___| |_ ___  _ __ ___ 
#  | |    / _ \| '_ \| '_ \ / _ \/ __| __/ _ \| '__/ __|
#  | |___| (_) | | | | | | |  __/ (__| || (_) | |  \__ \
#   \_____\___/|_| |_|_| |_|\___|\___|\__\___/|_|  |___/
#

# CONNECTORS specify what data will be pulled into the in-memory database during processing

connectors:

# Configure HTTP & Salesforce credentials in your Environment before running this recipe
  # HTTP Setup - https://help.lingk.io/en/articles/304-http-connector-setup-guide
  # Salesforce Setup - https://help.lingk.io/en/articles/267-salesforce-connector-setup-guide

###### Start - HTTP Reader connectors ######

## PeopleSoft Person Connector
  ## HTTP V2 connector in comparison with V1, handles pagination through the connector properties
  ## Authentication and URL values are provided via environment variables  
    # documentation - https://help.lingk.io/en/articles/202-custom-environment-variables

## PeopleSoft Education History Connector
- 
    name: peopleSoftEducationData
    type: httpV2Reader 
    properties:
      httpMethod: POST 
      headerJson: {
          "Authorization":"Basic {{env.vars.apiAccessToken}}",
         "Content-Type":"application/json"
        }
      acceptInvalidCerts: false
      url: '{{env.vars.midStateInstanceURL}}'
      paginationType: requestBody          
      body: 
                {
                  "isDebugMode": true,
                  "rowLimit": 1000,
                  "pageNumber": 1,
                  "noEffectiveDateLogic": false,
                  "noEffectiveStatusLogic": false,
                  "includeFieldTypes": false,
                  "records": [
                    {
                      "recordName": "PERSON",
                      "sqlWhereClause": " EXISTS (SELECT 1 FROM PS_SALFM_ADM_STDT B WHERE A.EMPLID = B.EMPLID AND B.INSTITUTION = '{{env.inst-code}}') and EXISTS (SELECT 1 FROM PS_EXT_DEGREE B WHERE B.EMPLID = A.EMPLID)"
                    },
                    {
                      "recordName": "EXT_ACAD_DATA",
                      "parentRecordName": "PERSON"
                    },
                    {
                      "recordName": "EXT_ACAD_SUM",
                      "parentRecordName": "EXT_ACAD_DATA"
                    },
                    {
                      "recordName": "EXT_DEGREE",
                      "parentRecordName": "PERSON",
                      "includeDescriptionsFor": ["DEGREE", "DEGREE_STATUS", "LS_DATA_SOURCE"]
                    }
                  ]
                }
    schema:
      fields:
        - name: 'BIRTHCOUNTRY'
          type: 'string'
        - name: 'BIRTHDATE'
          type: 'string'
        - name: 'BIRTHPLACE'
          type: 'string'
        - name: 'BIRTHSTATE'
          type: 'string'
        - name: 'DT_OF_DEATH'
          type: 'string'
        - name: 'EMPLID'
          type: 'string'
        - name: 'EXT_ACAD_DATA'
          type: 'object'
          objectSchema:
            name: objectSchemaSchema
            fields:
              - name: 'fields'
                type: 'array'
                items:
                  name: 'arrayElement'
                  type: 'object'
                  objectSchema:
                    name: objectSchemaSchema
                    fields:
                      - name: 'ACAD_LOAD_APPR'
                        type: 'string'
                      - name: 'DATA_MEDIUM'
                        type: 'string'
                      - name: 'EMPLID'
                        type: 'string'
                      - name: 'EXT_ACAD_LEVEL'
                        type: 'string'
                      - name: 'EXT_ACAD_SUM'
                        type: 'object'
                        objectSchema:
                          name: objectSchemaSchema
                          fields:
                            - name: 'fields'
                              type: 'array'
                              items:
                                name: 'arrayElement'
                                type: 'object'
                                objectSchema:
                                  name: objectSchemaSchema
                                  fields:
                                    - name: 'EMPLID'
                                      type: 'string'
                            - name: 'objectName'
                              type: 'string'
                            - name: 'objectType'
                              type: 'string'
                      - name: 'EXT_CAREER'
                        type: 'string'
                      - name: 'EXT_DATA_NBR'
                        type: 'long'
                      - name: 'EXT_ORG_ID'
                        type: 'string'
                      - name: 'EXT_TERM'
                        type: 'string'
                      - name: 'EXT_TERM_TYPE'
                        type: 'string'
                      - name: 'FROM_DT'
                        type: 'string'
                      - name: 'LS_DATA_SOURCE'
                        type: 'string'
                      - name: 'RECEIVED_DT'
                        type: 'string'
                      - name: 'SAD_DISMISSED'
                        type: 'string'
                      - name: 'SCC_ROW_ADD_DTTM'
                        type: 'string'
                      - name: 'SCC_ROW_ADD_OPRID'
                        type: 'string'
                      - name: 'SCC_ROW_UPD_DTTM'
                        type: 'string'
                      - name: 'SCC_ROW_UPD_OPRID'
                        type: 'string'
                      - name: 'TERM_YEAR'
                        type: 'long'
                      - name: 'TO_DT'
                        type: 'string'
                      - name: 'TRANSCRIPT_DT'
                        type: 'string'
                      - name: 'TRANSCRIPT_FLAG'
                        type: 'string'
                      - name: 'TRANSCRIPT_TYPE'
                        type: 'string'
                      - name: 'TRNSCRPT_STATUS'
                        type: 'string'
              - name: 'objectName'
                type: 'string'
              - name: 'objectType'
                type: 'string'
        - name: 'EXT_DEGREE'
          type: 'object'
          objectSchema:
            name: objectSchemaSchema
            fields:
              - name: 'fields'
                type: 'array'
                items:
                  name: 'arrayElement'
                  type: 'object'
                  objectSchema:
                    name: objectSchemaSchema
                    fields:
                      - name: 'DEGREE'
                        type: 'string'
                      - name: 'DEGREE_DT'
                        type: 'string'
                      - name: 'DEGREE_STATUS'
                        type: 'string'
                      - name: 'DEGREE_STATUS_description'
                        type: 'string'
                      - name: 'DEGREE_description'
                        type: 'string'
                      - name: 'DESCR'
                        type: 'string'
                      - name: 'EMPLID'
                        type: 'string'
                      - name: 'EXT_CAREER'
                        type: 'string'
                      - name: 'EXT_DATA_NBR'
                        type: 'long'
                      - name: 'EXT_DEGREE_NBR'
                        type: 'long'
                      - name: 'EXT_ORG_ID'
                        type: 'string'
                      - name: 'EXT_SUBJ_AREA_1'
                        type: 'string'
                      - name: 'EXT_SUBJ_AREA_2'
                        type: 'string'
                      - name: 'FIELD_OF_STUDY_1'
                        type: 'string'
                      - name: 'FIELD_OF_STUDY_2'
                        type: 'string'
                      - name: 'HONORS_CATEGORY'
                        type: 'string'
                      - name: 'LS_DATA_SOURCE'
                        type: 'string'
                      - name: 'LS_DATA_SOURCE_description'
                        type: 'string'
              - name: 'objectName'
                type: 'string'
              - name: 'objectType'
                type: 'string'
        - name: 'LAST_CHILD_UPDDTM'
          type: 'string'
        - name: 'rowNumber'
          type: 'long'

###### End - HTTP Reader connectors ######

###### Start - Salesforce Reader connectors ######

-
    name: sfContactReader
    type: salesforceReader
    delayedRead: true
    parameterizedBy: batchedStudentId
    properties:
      useBearerToken: true
      query: 
        SELECT 
          Id,
          University_Student_ID__c  
        FROM Contact
        WHERE University_Student_ID__c IN (
            {% for element in var.output %}
            '{{element.id}}'{% if not loop.last %},{% endif %}
            {% endfor %}
          ) 
    schema:
      fields:
        - name: 'Id'
          type: 'string'
        - name: 'University_Student_ID__c'
          type: 'string'

-
    name: sfAccountReader
    type: salesforceReader
    delayedRead: true
    parameterizedBy: batchedAccountId
    properties:
      useBearerToken: true
      query: 
        SELECT 
          Id,
          PeopleSoft_External_Org_ID__c  
        FROM Account
        WHERE PeopleSoft_External_Org_ID__c IN (
            {% for element in var.output %}
            '{{element.id}}'{% if not loop.last %},{% endif %}
            {% endfor %}
          ) 
    schema:
      fields:
        - name: 'Id'
          type: 'string'
        - name: 'PeopleSoft_External_Org_ID__c'
          type: 'string'

-
    name: sfEducationHistoryReader
    type: salesforceReader
    delayedRead: true
    parameterizedBy: batchedEducationId
    properties:
      useBearerToken: true
      query: 
        SELECT 
          Id,
          hed__Account__c,
          hed__Contact__c,
          hed__End_Date__c,
          hed__Degree_Earned__c,
          hed__Graduation_Date__c,
          PeopleSoft_ACAD_HISTORY_EXT_ORG_ID__c 
        FROM hed__Education_History__c
        WHERE PeopleSoft_ACAD_HISTORY_EXT_ORG_ID__c IN (
            {% for element in var.output %}
            '{{element.id}}'{% if not loop.last %},{% endif %}
            {% endfor %}
          ) 
    schema:
      fields:
        - name: 'Id'
          type: 'string'
        - name: 'hed__Account__c'
          type: 'string'
        - name: 'hed__Contact__c'
          type: 'string'
        - name: 'hed__End_Date__c'
          type: 'string'
        - name: 'hed__Degree_Earned__c'
          type: 'string'
        - name: 'hed__Graduation_Date__c'
          type: 'string'
        - name: 'PeopleSoft_ACAD_HISTORY_EXT_ORG_ID__c'
          type: 'string'

###### End - Salesforce Reader connectors ######

###### Start - Salesforce Writer connectors ######

-
    name: sfEducationHistoryWriter
    type: salesforceBulkWriter
    onError: continue
    isSerial: true
    properties:
      onError: continue
      useBearerToken: true
      sfObject: hed__Education_History__c
      batchSize: 100

###### End - Salesforce Writer connectors ######

###### Start - Local Writer connectors ######

# Local File Writer
-   
    name: finalMappedRecordFile
    type: localFileWriter
    format: csvWriter
    properties:
      fileName: "finalMappedEducationHistoryRecords.csv"

###### End - Local Writer connectors ######

#   _______        _        
#  |__   __|      | |       
#     | | __ _ ___| | _____ 
#     | |/ _` / __| |/ / __|
#     | | (_| \__ \   <\__ \
#     |_|\__,_|___/_|\_\___/

# Batch task used for parameterized by statements/connectors

tasks:
-
    name: batchTask
    type: dataOperation 
    function: batch   
    parameters:
       inputBatchGroupBy: # required
         id: desc      
       inputBatchSize: 200 # inputBatchSize # required (to test)
       inputBatchFields: "id" # comma delimited for other fields #inputBatchFields #required
       batchedColumnAlias: output # outputColumnName - optional with a default of "output"
       # inputTable: # optional
       # result: #optional

#   ______                         _       
#  |  ____|                       | |      
#  | |__ ___  _ __ _ __ ___   __ _| |_ ___ 
#  |  __/ _ \| '__| '_ ` _ \ / _` | __/ __|
#  | | | (_) | |  | | | | | | (_| | |_\__ \
#  |_|  \___/|_|  |_| |_| |_|\__,_|\__|___/

readFormats:
-
    name: text
    type: text

writeFormats:
-   
    name: csvWriter
    type: delimited
    properties:
      quoteAllFields: true
      delimiter: ','
      header: true

#    _____ _        _                            _       
#   / ____| |      | |                          | |      
#  | (___ | |_ __ _| |_ ___ _ __ ___   ___ _ __ | |_ ___ 
#   \___ \| __/ _` | __/ _ \ '_ ` _ \ / _ \ '_ \| __/ __|
#   ____) | || (_| | ||  __/ | | | | |  __/ | | | |_\__ \
#  |_____/ \__\__,_|\__\___|_| |_| |_|\___|_| |_|\__|___/

# STATEMENTS specify how the data should be processed while in memory

statements:

#******************************************************************** D I S C L A I M E R ***********************************************************************************************
#                                                                                                                                                                                       * 
# Note that in an effort to keep recipes optimized for DPH (Data Processing Hours), print statements should be commented out after development has concluded for a recipe.              *
# For more information on DPH optimization, please visit the following help article - https://help.lingk.io/en/articles/212-minimizing-data-processing-hours-on-the-lingk-platform      *
#                                                                                                                                                                                       * 
#******************************************************************** D I S C L A I M E R ***********************************************************************************************

#- statement: print peopleSoftEducationData

# - statement: (inlinePersonData) => SELECT inline(data.PERSON.fields) FROM peopleSoftEducationData
  # - statement: PRINT inlinePersonData

# inline function - flatten the array to make it a basic table
  - statement: (inlineAcadData) => SELECT inline(EXT_ACAD_DATA.fields) FROM peopleSoftEducationData
  #- statement: PRINT inlineAcadData

# inline function - flatten the array to make it a basic table
  - statement: (inlineDegreeData) => SELECT inline(EXT_DEGREE.fields) FROM peopleSoftEducationData
  #- statement: PRINT inlineDegreeData

# inline function - flatten the array to make it a basic table
  - statement: (inlineAcademicSum) => SELECT *,inline(EXT_ACAD_SUM.fields) FROM inlineAcadData
  #- statement: PRINT inlineAcademicSum

## Using the ROW_NUMBER function assigns a unique, sequential number to each row, starting with one,
   # according to the ordering of rows within the partition from the information pulled in the previous statement for the EXT_ACAD_SUM
  ## The PARTITION BY is dividing the rows into small partitions by TO_DT preference field
  - statement: (academicDataGroupByEmplIdAndOrgId) =>
                  SELECT
                    *,
                    ROW_NUMBER() OVER (PARTITION BY TRIM(EMPLID), TRIM(EXT_ORG_ID) ORDER BY TRIM(TO_DT) DESC NULLS LAST) AS Row
                  FROM inlineAcadData
  #- statement: PRINT academicDataGroupByEmplIdAndOrgId

## Using the ROW_NUMBER function assigns a unique, sequential number to each row, starting with one,
   # according to the ordering of rows within the partition from the information pulled in the previous statement for the EXT_DEGREE
  ## The PARTITION BY is dividing the rows into small partitions by DEGREE_DT preference field
  - statement: (degreeGroupByEmplIdOrgIdAndDegree) =>
                  SELECT
                    *,
                    ROW_NUMBER() OVER (PARTITION BY TRIM(EMPLID), TRIM(EXT_ORG_ID), TRIM(EXT_DEGREE_NBR) ORDER BY TRIM(DEGREE_DT) DESC) AS Row
                  FROM inlineDegreeData
  #- statement: PRINT degreeGroupByEmplIdOrgIdAndDegree

- statement: (finalInlineDegree) => SELECT * FROM degreeGroupByEmplIdOrgIdAndDegree WHERE Row = 1
  #- statement: PRINT finalInlineDegree
  - statement: (finalInlineDegreeCount) => SELECT COUNT(*) FROM finalInlineDegree
  #- statement: PRINT finalInlineDegreeCount

- statement: (distinctEmplId) => SELECT DISTINCT IFNULL(TRIM(EMPLID), '') `id` FROM peopleSoftEducationData

# This statement executes a batch process task where the name, inputTable, outputTable and the results are the parameters needed 
  - statement: |
        execute task 
            --name batchTask
            --inputTable distinctEmplId
            --outputBatchTable batchedStudentId
            --result resultStatus

- statement: REFRESH sfContactReader
  - statement: (sfContactReaderCount) => SELECT COUNT(*) FROM sfContactReader
  #- statement: PRINT sfContactReaderCount

# The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table)
  - statement: (nonMatchingContact) => SELECT
                                        *
                                       FROM finalInlineDegree d
                                       LEFT ANTI JOIN sfContactReader sf ON TRIM(d.EMPLID) = TRIM(sf.University_Student_ID__c)
  #- statement: PRINT nonMatchingContact
  - statement: (nonMatchingContactCount) => SELECT COUNT(*) FROM nonMatchingContact
  #- statement: PRINT nonMatchingContactCount

- statement: (distinctOrgId) => SELECT DISTINCT IFNULL(TRIM(EXT_ORG_ID), '') `id` FROM inlineDegreeData

# This statement executes a batch process task where the name, inputTable, outputTable and the results are the parameters needed 
  - statement: |
        execute task 
            --name batchTask
            --inputTable distinctOrgId
            --outputBatchTable batchedAccountId
            --result resultStatus

- statement: REFRESH sfAccountReader
  - statement: (sfAccountReaderCount) => SELECT COUNT(*) FROM sfAccountReader
  #- statement: PRINT sfAccountReaderCount

# The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table)
  - statement: (nonMatchingAccount) => SELECT
                                        *
                                       FROM finalInlineDegree d
                                       LEFT ANTI JOIN sfAccountReader sf ON TRIM(d.EXT_ORG_ID) = TRIM(sf.PeopleSoft_External_Org_ID__c)
  #- statement: PRINT nonMatchingAccount
  - statement: (nonMatchingAccountCount) => SELECT COUNT(*) FROM nonMatchingAccount
  #- statement: PRINT nonMatchingAccountCount

- statement: (rejectedRecordDueToBlankFields) =>
                  SELECT
                    *
                  FROM finalInlineDegree
                  WHERE (TRIM(EMPLID) = '' OR TRIM(EMPLID) = NULL)
                    AND (TRIM(EXT_ORG_ID) = '' OR TRIM(EXT_ORG_ID) = NULL)
                    AND (TRIM(EXT_DEGREE_NBR) = '' OR TRIM(EXT_DEGREE_NBR) = NULL)
  #- statement: PRINT rejectedRecordDueToBlankFields
  - statement: (rejectedRecordDueToBlankFieldsCount) => SELECT COUNT(*) FROM rejectedRecordDueToBlankFields
  #- statement: PRINT rejectedRecordDueToBlankFieldsCount

# CASE function works as a dictionary and retrieve a specific value depending to the type field value pulled from the table as long as it is not null.
  # The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table)
  - statement: |
      (mappedEducation) => 
          SELECT
            ac.Id                                                                                         `hed__Account__c`,
            c.Id                                                                                          `hed__Contact__c`,
            IFNULL(TRIM(to_date(a.TO_DT)), '')                                                            `hed__End_Date__c`,
            CASE WHEN IFNULL(TRIM(UPPER(d.DEGREE)), '') = 'HSD'
                 THEN 'High School Diploma'
                 WHEN IFNULL(TRIM(UPPER(d.DEGREE)), '') = 'HSED'
                 THEN 'High School Equivalency Diploma'
            ELSE IFNULL(TRIM(d.DEGREE), '')
            END                                                                                           `hed__Degree_Earned__c`,
            IFNULL(TRIM(to_date(d.DEGREE_DT)), '')                                                        `hed__Graduation_Date__c`,
            concat_ws('-', TRIM(d.EMPLID), TRIM(d.EXT_ORG_ID), TRIM(d.EXT_DEGREE_NBR))                    `PeopleSoft_ACAD_HISTORY_EXT_ORG_ID__c`

FROM finalInlineDegree d
          INNER JOIN sfContactReader c ON IFNULL(TRIM(c.University_Student_ID__c), '') = IFNULL(TRIM(d.EMPLID), '')
          INNER JOIN sfAccountReader ac ON IFNULL(TRIM(ac.PeopleSoft_External_Org_ID__c), '') = IFNULL(TRIM(d.EXT_ORG_ID), '')
          LEFT JOIN academicDataGroupByEmplIdAndOrgId a ON IFNULL(TRIM(d.EMPLID), '') = IFNULL(TRIM(a.EMPLID), '') 
            AND IFNULL(TRIM(d.EXT_ORG_ID), '') = IFNULL(TRIM(a.EXT_ORG_ID), '')
            AND a.Row = 1
          WHERE TRIM(d.EMPLID) != '' AND TRIM(d.EXT_ORG_ID) != '' AND TRIM(d.EXT_DEGREE_NBR) != ''

#- statement: print mappedEducation        
  - statement: (mappedEducationCount) => SELECT COUNT(*) FROM mappedEducation
  #- statement: print mappedEducationCount

- statement: (distinctEducationHistoryId) => SELECT DISTINCT IFNULL(TRIM(PeopleSoft_ACAD_HISTORY_EXT_ORG_ID__c), '') `id` FROM mappedEducation

# This statement executes a batch process task where the name, inputTable, outputTable and the results are the parameters needed 
  - statement: |
        execute task 
            --name batchTask
            --inputTable distinctEducationHistoryId
            --outputBatchTable batchedEducationId
            --result resultStatus

- statement: REFRESH sfEducationHistoryReader
  - statement: (sfEducationHistoryReaderCount) => SELECT COUNT(*) FROM sfEducationHistoryReader
  #- statement: print sfEducationHistoryReaderCount

- statement:
      (educationToUpdate) =>
          SELECT
            sf.Id,
            m.hed__Account__c,
            m.hed__Contact__c,
            m.hed__End_Date__c,
            m.hed__Degree_Earned__c,
            m.hed__Graduation_Date__c

FROM mappedEducation m
          INNER JOIN sfEducationHistoryReader sf ON IFNULL(TRIM(m.PeopleSoft_ACAD_HISTORY_EXT_ORG_ID__c), '') = IFNULL(TRIM(sf.PeopleSoft_ACAD_HISTORY_EXT_ORG_ID__c), '')

#- statement: print educationToUpdate
  - statement: (educationToUpdateCount) => SELECT COUNT(*) FROM educationToUpdate
  #- statement: print educationToUpdateCount

# The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table)
  - statement:
      (educationToInsert) =>
          SELECT
            m.*
          FROM mappedEducation m
          LEFT ANTI JOIN sfEducationHistoryReader sf ON IFNULL(TRIM(m.PeopleSoft_ACAD_HISTORY_EXT_ORG_ID__c), '') = IFNULL(TRIM(sf.PeopleSoft_ACAD_HISTORY_EXT_ORG_ID__c), '')

#- statement: print educationToInsert
  - statement: (educationToInsertCount) => SELECT COUNT(*) FROM educationToInsert
  #- statement: print educationToInsertCount

- statement: (updateResult, updateError) => UPDATE educationToUpdate INTO sfEducationHistoryWriter
  #- statement: print updateResult
  #- statement: print updateError

- statement: (insertResult, insertError) => INSERT educationToInsert INTO sfEducationHistoryWriter
  #- statement: print insertResult
  #- statement: print insertError

#################################### LOGS - Debugging purpuses ######################################

#  - statement: PRINT peopleSoftEducationDataCount
#  - statement: PRINT finalInlineDegreeCount
#  - statement: print sfContactReaderCount
#  - statement: PRINT nonMatchingContactCount
#  - statement: print sfAccountReaderCount
#  - statement: PRINT nonMatchingAccountCount
#  - statement: PRINT rejectedRecordDueToBlankFieldsCount
#  - statement: print mappedEducationCount
#  - statement: print sfEducationHistoryReaderCount
#  - statement: print educationToUpdateCount
#  - statement: print educationToInsertCount

###### Start - Insert Local file Contacts ######

- statement: REFRESH sfEducationHistoryReader

# The 'LEFT ANTI JOIN' function retrieves the rows from the first table (table after the FROM statement) that do not match with the rows on the second table (right table)
  - statement: |
      (existingEducationhistoryInfo) =>
          SELECT
            sf.Id,
            m.*
          FROM mappedEducation m
          LEFT JOIN sfEducationHistoryReader sf ON IFNULL(TRIM(m.PeopleSoft_ACAD_HISTORY_EXT_ORG_ID__c), '') = IFNULL(TRIM(sf.PeopleSoft_ACAD_HISTORY_EXT_ORG_ID__c), '')

- statement: INSERT existingEducationhistoryInfo INTO finalMappedRecordFile

###### End - Insert Local file Contacts ######

# Add more statements to convert, join, aggregrate, transform, and integrate your data