我有一个从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
}
请说明发生这种情况的原因。
暂无答案!
目前还没有任何答案,快来回答吧!