scala—functionmapping之后aws表中更改的记录数

我有一个从aws数据源获取记录的代码。一个函数Map到它之后，记录的数量突然下降。下面的代码给出了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"
    ))

在我将一个函数Map到这个数据源之后，记录的数量降到了1800万条。

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

Map的函数如下所示

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
    }

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

scala—functionmapping之后aws表中更改的记录数

暂无答案！

相关问题

热门标签

最新问答