FunctionMapping之后，AWS表中的记录数发生了变化

如何解决FunctionMapping之后，AWS表中的记录数发生了变化

我有一个从AWS数据源获取记录的代码。没有功能映射到记录后，记录数量突然下降。以下代码提供了1900万条记录。

val datasource0 = 
  glueContext
    .getCatalogSource(
      database = "apex-raw",tableName = "ownermaster",redshiftTmpDir = "",transformationContext="datasource0"
    )
    .getDynamicFrame()
    .selectFields(Seq(
      "ownermasterid","companyname","dateofbirth","firstname","middlename","lastname","homephone","homephoneextension","workphone","workphoneextension","cellphone","cellphoneextension","mailingaddressindicator","legaladdressindicator","addressline1","addressline2","addressline3","apartmentnumber","city","state","zipcode","zipcodeextension","addresslocationcode","countrycode","taxidnumber","taxidtype","emailaddress","lastupdated","lastupdatedby","maritalstatus"
    ))

将函数映射到此数据源后，没有。的记录下降到1800万。

val datasource0map = 
  datasource0
    .map(
      f = checkEmpty,errorMsg = "emptystringremains",transformationContext = "checkEmptyDataContext"
    )

映射的函数在下面给出

def checkEmpty(dynamicRecord: DynamicRecord): DynamicRecord = {
    
    val utaxIdNumber = dynamicRecord.getField("taxIdNumber").mkString("").trim
    val utaxIdType = dynamicRecord.getField("taxidtype").mkString("").trim
    val md = MessageDigest.getInstance("MD5")
    val s = dynamicRecord.getField("entityId").mkString("")
    val digest = md.digest(s.getBytes)
    val bigInt = new BigInteger(1,digest)
    val hashedString = bigInt.toString(16)
    val s_uuid = utaxIdType+utaxIdNumber
    
    val scomp = dynamicRecord.getField("businessLegalName").mkString("")
    val etype = if(scomp == null || scomp.trim.isEmpty){
        
    
    val taxId = utaxIdNumber.toLong.toString
    val ftaxIdNumber = utaxIdNumber.replaceFirst("(\\w{3})(\\w{2})(\\w+)","$1-$2-$3")
    dynamicRecord.dropField("taxIdNumber")
    dynamicRecord.addField("taxIdNumber",StringNode(ftaxIdNumber))
    dynamicRecord.addField("taxIdNumberOrig",StringNode(taxId))
    "Individual"
     } else {
            
            val taxId = utaxIdNumber.toLong.toString
            val ftaxIdNumber = utaxIdNumber.replaceFirst("(\\w{2})(\\w+)","$1-$2")
             dynamicRecord.dropField("taxIdNumber")
            dynamicRecord.addField("taxIdNumber",StringNode(ftaxIdNumber))
            dynamicRecord.addField("taxIdNumberOrig",StringNode(taxId))
    "Entity"
        }
    dynamicRecord.addField("entityType",StringNode(etype))
    dynamicRecord.addField("PhAddId",StringNode(hashedString))
    dynamicRecord.addField("uuid",StringNode(s_uuid))
    dynamicRecord.dropField("entityId")
    dynamicRecord.addField("entityId",StringNode(s))
    dynamicRecord.dropField("taxidtype")
    dynamicRecord.addField("taxIdType",StringNode(utaxIdType))
    
    val s_home = dynamicRecord.getField("homephone").mkString("")
    val s_homeextn = dynamicRecord.getField("homephoneextension").mkString("")
    val s_cell = dynamicRecord.getField("cellphone").mkString("")
    val s_cellextn = dynamicRecord.getField("cellphoneextension").mkString("")
    val s_work = dynamicRecord.getField("workphone").mkString("")
    val s_workextn = dynamicRecord.getField("workphoneextension").mkString("")
    
    val s_phone = if(s_home != null || !s_home.trim.isEmpty){
    s_home.trim
        } 
        else if(s_cell != null || !s_cell.trim.isEmpty){
            s_cell.trim
        }
        else{
            s_work.trim
        }
        
    val s_extn = if(s_home != null || !s_home.trim.isEmpty){
    s_homeextn
        } 
        else if(s_cell != null || !s_cell.trim.isEmpty){
            s_cellextn
        }
        else{
            s_workextn
        }

    dynamicRecord.addField("phonenumber",StringNode(s_phone))
    dynamicRecord.addField("phoneextension",StringNode(s_extn))
    dynamicRecord.addField("phonetypecode",StringNode(""))
    
    
    val s_maritalstatus = dynamicRecord.getField("maritalstatus").mkString("").trim
    val s_taxidtype = dynamicRecord.getField("taxidtype").mkString("").trim
    val s_firstName = dynamicRecord.getField("firstName").mkString("").trim
    val s_middleInitiale = dynamicRecord.getField("middleInitial").mkString("").trim
    val s_lastName = dynamicRecord.getField("lastName").mkString("").trim
    val s_BLName = dynamicRecord.getField("businessLegalName").mkString("").trim
    val s_email = dynamicRecord.getField("emailaddress").mkString("").trim
    val s_streetName = dynamicRecord.getField("addressline1").mkString("").trim
    val s_addressline2 = dynamicRecord.getField("addressline2").mkString("").trim
    val s_addressline3 = dynamicRecord.getField("addressline3").mkString("").trim
    val s_city = dynamicRecord.getField("city").mkString("").trim 
    val s_apartment = dynamicRecord.getField("apartmentnumber").mkString("").trim
    val s_mailingaddressindicator = dynamicRecord.getField("mailingaddressindicator").mkString("").trim 
    val s_legaladdressindicator = dynamicRecord.getField("legaladdressindicator").mkString("").trim 
    val s_zipcode = dynamicRecord.getField("zipcode").mkString("").trim
    val s_zipcodeextension = dynamicRecord.getField("zipcodeextension").mkString("").trim  
    val s_lastupdatedby = dynamicRecord.getField("LastUpdatedBy").mkString("").trim
    val s_country = dynamicRecord.getField("countrycode").mkString("").trim
    val s_state = dynamicRecord.getField("state").mkString("").trim
    val inc_type = if(s_mailingaddressindicator == "Y"){
        "MailingAddress"
        } else {
        "LegalAddress"
    }
     
     val zipext = if(s_zipcodeextension == null || s_zipcodeextension.trim.isEmpty){
        ""
        } else {
        "("+s_zipcodeextension+")"
    }
    val streetnameconcat = (s_addressline2+" "+s_addressline3).trim
    val zipcodeconcat = (s_zipcode+zipext).trim
    
    val addressidvalue = s_country.toUpperCase()+s_legaladdressindicator+s_city.toUpperCase()+s_mailingaddressindicator+inc_type.toUpperCase()+streetnameconcat.toUpperCase()+zipcodeconcat.toUpperCase()+s_apartment.toUpperCase()+s_streetName.toUpperCase()+s_state.toUpperCase()
    
    val fnameidvalue = s_firstName.toUpperCase()
    
    val lnameidvalue = s_lastName.toUpperCase()
    
    val digestadd = md.digest(addressidvalue.getBytes)
    val bigIntadd = new BigInteger(1,digestadd)
    val hashedStringadd = bigIntadd.toString(16)
    
    dynamicRecord.addField("addressidvalue",StringNode(hashedStringadd))
    
    val digestph = md.digest(s_phone.getBytes)
    val bigIntph = new BigInteger(1,digestph)
    val hashedStringph = bigIntph.toString(16)
    
    dynamicRecord.addField("phoneidvalue",StringNode(hashedStringph))
    
    val digestfn = md.digest(fnameidvalue.getBytes)
    val bigIntfn = new BigInteger(1,digestfn)
    val hashedStringfn = bigIntfn.toString(16)
    
    dynamicRecord.addField("fnameidvalue",StringNode(hashedStringfn))
    
    val digestln = md.digest(lnameidvalue.getBytes)
    val bigIntln = new BigInteger(1,digestln)
    val hashedStringln = bigIntln.toString(16)
    
    dynamicRecord.addField("lnameidvalue",StringNode(hashedStringln))
    
    
    dynamicRecord.dropField("firstName")
    dynamicRecord.addField("firstName",StringNode(s_firstName.toUpperCase()))
    
    dynamicRecord.dropField("state")
    dynamicRecord.addField("state",StringNode(s_state))
    
    dynamicRecord.dropField("middleInitial")
    dynamicRecord.addField("middleInitial",StringNode(s_middleInitiale.toUpperCase()))
    
    dynamicRecord.dropField("countrycode")
    dynamicRecord.addField("countrycode",StringNode(s_country.toUpperCase()))
    
    dynamicRecord.dropField("lastName")
    dynamicRecord.addField("lastName",StringNode(s_lastName.toUpperCase()))
    
    dynamicRecord.dropField("businessLegalName")
    dynamicRecord.addField("businessLegalName",StringNode(s_BLName))
    
    dynamicRecord.dropField("mailingaddressindicator")
    dynamicRecord.addField("mailingaddressindicator",StringNode(s_mailingaddressindicator))
    
    dynamicRecord.dropField("legaladdressindicator")
    dynamicRecord.addField("legaladdressindicator",StringNode(s_legaladdressindicator))
    
    dynamicRecord.dropField("emailaddress")
    dynamicRecord.addField("emailaddress",StringNode(s_email))
    
    dynamicRecord.dropField("addressline1")
    dynamicRecord.addField("streetName",StringNode(s_streetName))
    
    dynamicRecord.dropField("zipcode")
    dynamicRecord.addField("zipcode",StringNode(zipcodeconcat))
    dynamicRecord.dropField("addressline2")
    dynamicRecord.dropField("addressline3")
    
     dynamicRecord.dropField("apartmentnumber")
    dynamicRecord.addField("apartmentnumber",StringNode(s_apartment))
   
    dynamicRecord.dropField("homephone")
    dynamicRecord.addField("homephone",StringNode(s_home))
    dynamicRecord.dropField("workphone")
    dynamicRecord.addField("workphone",StringNode(s_work))
    dynamicRecord.dropField("cellphone")
    dynamicRecord.addField("cellphone",StringNode(s_cell))
    
    dynamicRecord.dropField("LastUpdatedBy")
    dynamicRecord.addField("LastUpdatedBy",StringNode(s_lastupdatedby))
  
    
    dynamicRecord.addField("streetName2",StringNode(streetnameconcat))

    dynamicRecord.dropField("city")
    dynamicRecord.addField("city",StringNode(s_city))
    dynamicRecord.addField("type",StringNode(inc_type))
    dynamicRecord.addField("order",StringNode("Primary"))

    dynamicRecord.dropField("maritalstatus")
    dynamicRecord.addField("maritalstatus",StringNode(s_maritalstatus))

    dynamicRecord.addField("entityKeyId",StringNode(utaxIdNumber))
    
    dynamicRecord
    }

请说明发生这种情况的任何原因。

FunctionMapping之后，AWS表中的记录数发生了变化

如何解决FunctionMapping之后，AWS表中的记录数发生了变化

相关推荐